diff --git a/docs/Historian.md b/docs/Historian.md index 214a59df5..cdf7a6d28 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -161,6 +161,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | Key | Type | Default | Description | |---|---|---|---| | `Gateway:EnableNativeTimeWeighted` | bool | `false` | Serve HistoryRead-Processed `TimeAverage` / `Total` through AVEVA's native retrieval modes. ⚠️ **Off by default (client-visible):** the gateway sends AVEVA the interval in the wrong unit (historiangw#27), so AVEVA answers one Good-status whole-span value on the earliest interval and nothing for the rest — wrong values a client cannot tell from right ones. Off, both are `BadAggregateNotSupported` and are **not advertised** (`AggregateFunctions`), like the other five time-weighted aggregates on this backend. Turn it on only after a fixed gateway is live-verified. | +| `Gateway:ProbeInterval` | TimeSpan | `00:00:30` | How often the read client calls the gateway's `ProbeAsync` for the historian health snapshot (`ProcessConnectionOpen` / `EventConnectionOpen`, shown on `/hosts`). The first probe runs at start-up, so a node that has served no HistoryRead yet — or only writes (continuous historization) — reports a healthy gateway as open; it used to read closed until some client happened to read. The latest observation (probe or read outcome) wins; probes never count as queries or failures. Zero/negative = no probe (read outcomes only, the previous behaviour). | | `Gateway:EventDrainBudget` | int | `10000` | Stream rows a capped HistoryReadEvents read consumes (and discards) past its page cap before it abandons the gateway stream. The gateway's `ReadEvents` has no max-events field, so the cap is applied client-side; an **abandoned** stream cancels the RPC and the gateway evicts its pooled historian session (the next read reconnects; concurrent reads on that session fail). The usual small overshoot is drained to the stream's end; past the budget the stream is abandoned, so a page over a window far larger than the page does not stream the whole window. Rows of another source never count before the "more events" signal is seen, so an abandoned stream never answers a page as complete. `0` = abandon at the cap (the previous behaviour); negative counts as `0` (a `Validate()` warning). | **`ServerHistorian:SqlServer`** (read only when `Backend=SqlServer`; bound by `HistorianBackendRegistry.BindSqlServer`): diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 1bf418e15..a34468f81 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -120,6 +120,24 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis private long _totalFailures; private int _consecutiveFailures; + /// + /// The latest connectivity observation — a read outcome or a probe result, whichever came last (null = none yet). + /// reports it as the connection flags. + /// + private bool? _lastObservationOpen; + + /// The probe cadence (ServerHistorian:Gateway:ProbeInterval); zero = no probe loop. + private readonly TimeSpan _probeInterval; + + /// The probe cadence — exposed so the factory wiring is testable. + internal TimeSpan ProbeInterval => _probeInterval; + + private readonly CancellationTokenSource _probeStop = new(); + private readonly Task _probeLoop = Task.CompletedTask; + + /// The most one probe may take before it counts as a failed (closed) observation. + private static readonly TimeSpan ProbeCallBound = TimeSpan.FromSeconds(10); + /// Creates a gateway-backed historian data source. /// The gateway client seam used for all reads. /// Diagnostic logger; failures are recorded without leaking tag/host detail. @@ -148,6 +166,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// default false): until historiangw#27 is live-verified they answer Good-status wrong values, so by default they /// are refused (BadAggregateNotSupported) and not advertised. /// + /// + /// Cadence of the ProbeAsync health loop (ServerHistorian:Gateway:ProbeInterval); the first probe runs + /// at construction. Null / zero (the default here; the factory passes the configured 30 s) = no loop, health from + /// read outcomes only. + /// /// /// Stream rows a capped events read consumes (uncollected) past its cap before it abandons the gateway stream /// (ServerHistorian:Gateway:EventDrainBudget). Negative counts as 0 = abandon at the cap. @@ -155,7 +178,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null, bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0, - int eventDrainBudget = DefaultEventDrainBudget, bool enableNativeTimeWeighted = false) + int eventDrainBudget = DefaultEventDrainBudget, bool enableNativeTimeWeighted = false, + TimeSpan? probeInterval = null) { _enableNativeTimeWeighted = enableNativeTimeWeighted; _eventDrainBudget = Math.Max(0, eventDrainBudget); @@ -170,6 +194,53 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis _time = timeProvider ?? TimeProvider.System; _ceiling = new GatewayRowCeiling(_time); _tags = new GatewayTagPresenceCache(_client, _time); + _probeInterval = probeInterval is { } pi && pi > TimeSpan.Zero ? pi : TimeSpan.Zero; + if (_probeInterval > TimeSpan.Zero) _probeLoop = Task.Run(() => ProbeLoopAsync(_probeStop.Token)); + } + + /// + /// Probes the gateway once immediately, then every , recording each result as the + /// latest connectivity observation. Probes are not history queries: they never touch the query / failure counters. + /// A probe that throws or overruns is a closed observation; nothing escapes the loop. + /// + private async Task ProbeLoopAsync(CancellationToken stop) + { + using var timer = new PeriodicTimer(_probeInterval, _time); + do + { + bool open; + try + { + using var bound = CancellationTokenSource.CreateLinkedTokenSource(stop); + bound.CancelAfter(ProbeCallBound); + open = await _client.ProbeAsync(bound.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stop.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + open = false; + _logger.LogDebug("Historian gateway probe failed ({Exception}).", ex.GetType().Name); + } + + bool changed; + lock (_healthLock) + { + changed = _lastObservationOpen != open; + _lastObservationOpen = open; + } + if (changed) + _logger.LogInformation("Historian gateway probe: the gateway is {State}.", open ? "reachable" : "unreachable"); + } + while (await WaitNextAsync(timer, stop).ConfigureAwait(false)); + } + + private static async Task WaitNextAsync(PeriodicTimer timer, CancellationToken stop) + { + try { return await timer.WaitForNextTickAsync(stop).ConfigureAwait(false); } + catch (OperationCanceledException) { return false; } } /// @@ -760,12 +831,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis LastSuccessTime: _lastSuccessUtc, LastFailureTime: _lastFailureUtc, LastError: _lastError, - // No connection probe exists (the old RefreshConnectionStateAsync had no production - // caller and was deleted, archreview 06/S-6): "open" is derived from observed reads - // alone -- at least one success and no failure since. The gateway is non-clustered to - // us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture). - ProcessConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0, - EventConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0, + // "Open" is the latest connectivity observation: a read outcome, or a ProbeAsync result from the probe + // loop (ServerHistorian:Gateway:ProbeInterval), whichever came last — so a node that has served no + // HistoryRead yet (or only writes) still reports a healthy gateway. Without a probe loop it is read + // outcomes alone: at least one success and no failure since. The gateway is non-clustered to us, so + // node fields are null/empty. + ProcessConnectionOpen: _lastObservationOpen ?? false, + EventConnectionOpen: _lastObservationOpen ?? false, ActiveProcessNode: null, ActiveEventNode: null, Nodes: []); @@ -850,6 +922,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis lock (_healthLock) { _totalQueries++; + _lastObservationOpen = success; if (success) { _totalSuccesses++; @@ -869,7 +942,16 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// Disposes the underlying gateway client. Prefer . public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult(); - /// Asynchronously disposes the underlying gateway client. + /// Stops the probe loop (bounded wait), then asynchronously disposes the underlying gateway client. /// A task that completes when the client has been disposed. - public ValueTask DisposeAsync() => _client.DisposeAsync(); + public async ValueTask DisposeAsync() + { + if (!_probeStop.IsCancellationRequested) + { + _probeStop.Cancel(); + // The loop never throws; a probe stuck past its own bound is abandoned rather than awaited further. + await Task.WhenAny(_probeLoop, Task.Delay(ProbeCallBound)).ConfigureAwait(false); + await _client.DisposeAsync().ConfigureAwait(false); + } + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index 98f7273ce..c3fcb6b83 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -50,7 +50,9 @@ public static class GatewayHistorian // How far a capped events read drains its stream past the cap before abandoning it. eventDrainBudget: options.Gateway.EventDrainBudget, // AVEVA's native TimeAverage / Total — off until historiangw#27 is live-verified. - enableNativeTimeWeighted: options.Gateway.EnableNativeTimeWeighted); + enableNativeTimeWeighted: options.Gateway.EnableNativeTimeWeighted, + // A slow ProbeAsync timer so the health snapshot reflects the gateway without waiting for a read. + probeInterval: options.Gateway.ProbeInterval); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index c44b9c67c..fce569120 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -291,4 +291,13 @@ public sealed class GatewayHistorianSettings /// time-weighted aggregates on this backend). Turn it on only once a fixed gateway is live-verified. /// public bool EnableNativeTimeWeighted { get; init; } + + /// + /// How often the read client probes the gateway (ProbeAsync) for the historian health snapshot's + /// connection flags. The first probe runs at start-up, so a node that has served no HistoryRead yet — or only + /// writes (continuous historization) — reports a healthy gateway as open instead of closed. The latest + /// observation (probe or read) wins. Default 30 s; zero or negative = no probe (health from read outcomes only, + /// the previous behaviour). + /// + public TimeSpan ProbeInterval { get; init; } = TimeSpan.FromSeconds(30); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayProbeHealthTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayProbeHealthTests.cs new file mode 100644 index 000000000..579dd86e0 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayProbeHealthTests.cs @@ -0,0 +1,129 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// The Gateway historian's connection health is no longer derived from read outcomes alone: a slow +/// ProbeAsync timer (ServerHistorian:Gateway:ProbeInterval) reports a healthy gateway as open on a node +/// that has served no HistoryRead yet (or only writes), and a dead one as closed. The latest observation — probe or +/// read — wins. +/// +public sealed class GatewayProbeHealthTests +{ + private static readonly TimeSpan Fast = TimeSpan.FromMilliseconds(30); + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static async Task Eventually(Func condition, string because) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException($"timed out waiting: {because}"); + await Task.Delay(10, Ct); + } + } + + [Fact] + public async Task A_healthy_gateway_reads_open_before_any_HistoryRead() + { + var fake = new FakeHistorianGatewayClient { ProbeResult = true }; + await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast); + + await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "the probe should open the connection flag"); + var h = ds.GetHealthSnapshot(); + h.EventConnectionOpen.ShouldBeTrue(); + h.TotalQueries.ShouldBe(0, "a probe is not a history query"); + fake.ProbeCallCount.ShouldBeGreaterThan(0); + } + + [Fact] + public async Task A_failing_or_throwing_probe_reads_closed_and_the_latest_observation_wins() + { + var fake = new FakeHistorianGatewayClient { ProbeResult = false }; + await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast); + await Eventually(() => fake.ProbeCallCount >= 2, "probes ran"); + ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse(); + + // A thrown probe (transport down) is closed too, and never kills the loop. + fake.ProbeThrows = new InvalidOperationException("down"); + var before = fake.ProbeCallCount; + await Eventually(() => fake.ProbeCallCount >= before + 2, "the loop survives a throwing probe"); + ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse(); + + // The probe recovers ⇒ open, with no read at all. + fake.ProbeThrows = null; + fake.ProbeResult = true; + await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "a recovered probe reopens"); + } + + [Fact] + public async Task A_failed_read_after_a_good_probe_closes_until_the_next_good_observation() + { + var fake = new FakeHistorianGatewayClient { ProbeResult = true, RawSamples = Array.Empty() }; + await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, + probeInterval: TimeSpan.FromHours(1)); + await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "the immediate first probe opens"); + + fake.ThrowOnRead = new InvalidOperationException("boom"); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", default, default, 1, Ct)); + ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse("the read failed after the probe"); + + fake.ThrowOnRead = null; + await ds.ReadRawAsync("T", default, default, 1, Ct); + ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeTrue(); + } + + [Fact] + public async Task Without_a_probe_interval_nothing_probes_and_health_stays_read_derived() + { + var fake = new FakeHistorianGatewayClient(); + await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + // Positive control in the same window: an enabled source does probe. + var control = new FakeHistorianGatewayClient(); + await using var probed = new GatewayHistorianDataSource(control, NullLogger.Instance, probeInterval: Fast); + + await Eventually(() => control.ProbeCallCount >= 2, "the control probes"); + fake.ProbeCallCount.ShouldBe(0); + ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse(); + } + + [Fact] + public async Task Disposing_stops_the_probe_loop() + { + var fake = new FakeHistorianGatewayClient(); + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast); + await Eventually(() => fake.ProbeCallCount >= 2, "probes ran"); + await ds.DisposeAsync(); + var after = fake.ProbeCallCount; + await Task.Delay(Fast * 5, Ct); + fake.ProbeCallCount.ShouldBe(after); + fake.DisposeCallCount.ShouldBe(1); + } + + [Theory] + [InlineData(null, 30)] + [InlineData("00:00:05", 5)] + [InlineData("00:00:00", 0)] + public void The_probe_interval_binds_from_configuration_and_reaches_the_data_source(string? configured, int expectedSeconds) + { + var values = new Dictionary + { + ["ServerHistorian:Enabled"] = "true", + ["ServerHistorian:Endpoint"] = "https://localhost:5222", + ["ServerHistorian:ApiKey"] = "histgw_x_y", + }; + if (configured is not null) values["ServerHistorian:Gateway:ProbeInterval"] = configured; + var opts = new ConfigurationBuilder().AddInMemoryCollection(values).Build() + .GetSection(ServerHistorianOptions.SectionName).Get()!; + using var services = new ServiceCollection().BuildServiceProvider(); + var ds = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); + ds.ProbeInterval.ShouldBe(TimeSpan.FromSeconds(expectedSeconds)); + ds.Dispose(); + } +}