Joseph Doherty
757 lines
44 KiB
C#
757 lines
44 KiB
C#
using System.Collections.Frozen;
|
||
using Grpc.Core;
|
||
using Microsoft.Extensions.Logging;
|
||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
|
||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||
|
||
/// <summary>
|
||
/// Server-side <see cref="IHistorianDataSource"/> backed by the HistorianGateway gRPC surface
|
||
/// (via the <see cref="IHistorianGatewayClient"/> seam). Translates OPC UA HistoryRead requests
|
||
/// to gateway read calls and maps the wire shapes back to the driver-agnostic
|
||
/// <see cref="DataValueSnapshot"/> / <see cref="HistoricalEvent"/> carriers using the pure
|
||
/// mappers in <c>Mapping/</c>.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// The data source owns no historian connection of its own — it delegates to the gateway, which
|
||
/// pools and amortizes the underlying historian sessions. A thrown gateway exception is recorded
|
||
/// as a health failure and rethrown: the node manager turns it into a Bad HistoryRead result, so
|
||
/// a backend fault never crashes the host. An empty time window is a successful (GoodNoData)
|
||
/// read, not a fault. Health counters follow the single-<c>_healthLock</c> discipline ported
|
||
/// from <c>WonderwareHistorianClient</c> so <c>TotalSuccesses + TotalFailures == TotalQueries</c>
|
||
/// holds at every observed snapshot.
|
||
/// </remarks>
|
||
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
|
||
{
|
||
/// <summary>
|
||
/// The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
|
||
/// aggregate set (#681): the per-sample aggregates evaluated from raw reads (every member of the
|
||
/// <see cref="AggregateFamily.Simple"/> / <see cref="AggregateFamily.StartEnd"/> /
|
||
/// <see cref="AggregateFamily.ActualTime"/> families of <see cref="AggregateFamilies"/>: Average, Count, Minimum,
|
||
/// Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
|
||
/// (TimeAverage, Total — <see cref="AggregateModeMapper"/>). Any other aggregate throws
|
||
/// <see cref="HistoryAggregateNotSupportedException"/> without calling the gateway. It is what
|
||
/// <see cref="SupportedAggregates"/> returns, so the advertised <c>AggregateFunctions</c> folders and the server's
|
||
/// up-front <c>BadAggregateNotSupported</c> read it too (#682).
|
||
/// </summary>
|
||
public static IReadOnlySet<HistoryAggregateType> ServedAggregates { get; } =
|
||
Enum.GetValues<HistoryAggregateType>()
|
||
.Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
|
||
.ToFrozenSet();
|
||
|
||
private readonly IHistorianGatewayClient _client;
|
||
private readonly ILogger<GatewayHistorianDataSource> _logger;
|
||
private readonly int _maxRawRows;
|
||
private readonly TimeProvider _time;
|
||
private readonly bool _atTimeFromRaw;
|
||
|
||
/// <summary>#693: AtTime from raw reads (true) or AVEVA's own interpolation (the kill switch).</summary>
|
||
internal bool AtTimeFromRaw => _atTimeFromRaw;
|
||
|
||
/// <summary>The most #687 crowding re-reads one raw read makes (#694 F4): each at least doubles the pre-window room.</summary>
|
||
private const int MaxCrowdedReReads = 8;
|
||
|
||
/// <summary>The default of <c>maxPreWindowRows</c>: <c>ServerHistorian:MaxTieClusterOverfetch</c>'s default.</summary>
|
||
public const int DefaultMaxPreWindowRows = 65536;
|
||
|
||
/// <summary>
|
||
/// The most rows before a raw window (inside the 1 ms look-back) a crowding re-read makes room for (#694 R2) — the
|
||
/// server's tie-cluster limit (<c>ServerHistorian:MaxTieClusterOverfetch</c>), which every other tie path treats as
|
||
/// the limit. More ⇒ <see cref="HistoryReadLimitExceededException"/> with <c>PreWindowRows</c> set.
|
||
/// </summary>
|
||
private readonly int _maxPreWindowRows;
|
||
|
||
/// <summary>The tie-cluster limit the crowding re-reads honour (#694 R2) — exposed so the factory wiring is testable.</summary>
|
||
internal int MaxPreWindowRows => _maxPreWindowRows;
|
||
|
||
/// <summary>A strict gateway's row ceiling, learned from its first refusal (#691.3).</summary>
|
||
private readonly GatewayRowCeiling _ceiling;
|
||
private readonly GatewayTagPresenceCache _tags;
|
||
|
||
private readonly object _healthLock = new();
|
||
private DateTime? _lastSuccessUtc;
|
||
private DateTime? _lastFailureUtc;
|
||
private string? _lastError;
|
||
private long _totalQueries;
|
||
private long _totalSuccesses;
|
||
private long _totalFailures;
|
||
private int _consecutiveFailures;
|
||
|
||
/// <summary>Creates a gateway-backed historian data source.</summary>
|
||
/// <param name="client">The gateway client seam used for all reads.</param>
|
||
/// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
|
||
/// <param name="maxRawRows">
|
||
/// In-window raw-row cap of one raw-evaluated processed read (<c>ServerHistorian:MaxTimeWeightedRawRows</c>;
|
||
/// <= 0 = unbounded). Past it the node fails <c>BadTooManyOperations</c>.
|
||
/// </param>
|
||
/// <param name="timeProvider">Clock for result server timestamps and the raw look-ahead horizon; null = system.</param>
|
||
/// <param name="atTimeFromRaw">
|
||
/// #693: serve AtTime from raw reads with the shared AtTime rules (<see cref="GatewayAtTimeReader"/>; default);
|
||
/// <c>false</c> = AVEVA's own interpolation passed through (<c>ServerHistorian:GatewayAtTimeFromRaw</c> kill switch).
|
||
/// </param>
|
||
/// <param name="maxPreWindowRows">
|
||
/// The most rows before a raw window a crowding re-read makes room for (#694 R2; <c>ServerHistorian:MaxTieClusterOverfetch</c>,
|
||
/// values below 1 count as 1). More ⇒ <see cref="HistoryReadLimitExceededException"/> (<c>BadTooManyOperations</c>).
|
||
/// </param>
|
||
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
|
||
int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null,
|
||
bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows)
|
||
{
|
||
_maxPreWindowRows = Math.Max(1, maxPreWindowRows);
|
||
_atTimeFromRaw = atTimeFromRaw;
|
||
ArgumentNullException.ThrowIfNull(client);
|
||
ArgumentNullException.ThrowIfNull(logger);
|
||
_client = client;
|
||
_logger = logger;
|
||
_maxRawRows = maxRawRows;
|
||
_time = timeProvider ?? TimeProvider.System;
|
||
_ceiling = new GatewayRowCeiling(_time);
|
||
_tags = new GatewayTagPresenceCache(_client, _time);
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// A reverse window (<c>start > end</c>) is refused with <see cref="NotSupportedException"/> before anything is sent
|
||
/// and is not recorded as a health failure (Gitea #686): the gateway rejects it <c>InvalidArgument</c>, and the server
|
||
/// never asks — <see cref="SupportsReverseRaw"/> is <c>false</c>, so it walks a reverse read back with forward reads.
|
||
/// Stored samples only (#687 — Part 11 Raw): AVEVA's synthetic start-boundary row is never returned, see
|
||
/// <see cref="ReadRawWindowAsync"/>.
|
||
/// </remarks>
|
||
public async Task<HistoryReadResult> ReadRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
if (startUtc > endUtc)
|
||
throw new NotSupportedException("The HistorianGateway reads raw history oldest-first only (start > end).");
|
||
|
||
try
|
||
{
|
||
if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult([], ContinuationPoint: null);
|
||
}
|
||
var samples = await ReadRawWindowAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
var snapshots = SampleMapper.ToSnapshots(samples);
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(snapshots, ContinuationPoint: null);
|
||
}
|
||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||
{
|
||
throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault
|
||
}
|
||
catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
|
||
{
|
||
// An unbounded read over Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false: a
|
||
// client-request limit (BadTooManyOperations), not a backend fault.
|
||
throw new HistoryReadLimitExceededException(maxValuesPerNode,
|
||
$"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " +
|
||
"Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window.");
|
||
}
|
||
catch (HistoryReadLimitExceededException)
|
||
{
|
||
// A client-request limit (#687: the pre-window margin swallowed the row budget), not a backend fault.
|
||
throw;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// The stored samples of <c>[startUtc, endUtc)</c> (a point read <c>start == end</c>: of that instant), oldest first,
|
||
/// at most <paramref name="maxValuesPerNode"/> (<c>0</c> = all) — without AVEVA's synthetic rows (#687).
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// An AVEVA <c>Full</c> raw read starts with a row stamped exactly at the query start: the value in effect there
|
||
/// (<c>Quality = 133</c>, the prior sample's value — never stored at that time) or, before the tag's first sample,
|
||
/// a NULL row (value 0, Bad). Part 11 Raw with <c>ReturnBounds = false</c> wants stored values only, and the
|
||
/// server's paging (#669) counts every returned row toward the page and resumes at the last returned timestamp,
|
||
/// so a fabricated row would also be counted as a sample and re-fabricated at every resume point. The read
|
||
/// therefore uses the same request start as the raw-evaluated aggregates (#681): it is requested from
|
||
/// <see cref="RawAggregateReader.RequestStartFor"/> (1 ms early) and only rows
|
||
/// <see cref="RawAggregateReader.IsWindowSample"/> are kept — the decision is by TIMESTAMP, because a stored NULL
|
||
/// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). The
|
||
/// forward end is exclusive (the interface contract; the gateway's own end is inclusive), a point read inclusive.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Row budget.</b> <c>0</c> ("all values") asks for <see cref="int.MaxValue"/> (the client package and the
|
||
/// gateway reject <c>max_values <= 0</c>, #686). A cap asks for <c>cap + </c>
|
||
/// <see cref="RawAggregateReader.PreWindowAllowance"/> rows — NOT clamped at the gateway's default ceiling (#688
|
||
/// clamps only the per-sample aggregate budget): AVEVA always emits the boundary row, so a clamped budget would make
|
||
/// every page size within the allowance of the ceiling come back short. A strict gateway
|
||
/// (<c>AllowStreamingBeyondCeiling=false</c>) refuses an over-ceiling budget and names its ceiling; the read is
|
||
/// then re-issued ONCE with the ceiling as budget (#691.3, <see cref="GatewayReadLimits"/>) — so an uncapped read
|
||
/// of a small window is served — and a ceiling-bounded read that is not provably complete (it delivered the
|
||
/// whole ceiling without reaching the window end, or a page short of its cap) fails
|
||
/// <see cref="HistoryReadLimitExceededException"/> (<c>BadTooManyOperations</c>), never a truncated answer.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Completeness.</b> A full page must hold exactly the cap of stored samples — a short page ends the client's
|
||
/// paging. A read is complete when the stream came back short, filled the cap, or reached the window end. If rows
|
||
/// BEFORE the window (the boundary row and stored samples inside the 1 ms margin, e.g. a tie cluster at
|
||
/// <c>start − 1 ms</c>) used up the budget, the window is read ONCE more asking for <c>cap + m + </c>
|
||
/// <see cref="RawAggregateReader.PreWindowAllowance"/> (m = the pre-window rows the first read saw); only if that
|
||
/// re-read is still short because of pre-window rows does the read fail with
|
||
/// <see cref="HistoryReadLimitExceededException"/>. Every stream is consumed to completion (never broken out of —
|
||
/// a cancelled gateway stream evicts its pooled historian session). <c>ReturnBounds</c> is not plumbed to the
|
||
/// data source on any backend, so bounding values are never synthesised here.
|
||
/// </para>
|
||
/// </remarks>
|
||
private async Task<List<HistorianSample>> ReadRawWindowAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
// The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range.
|
||
var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue);
|
||
if (cap == 0)
|
||
{
|
||
var all = await PassAsync(int.MaxValue).ConfigureAwait(false);
|
||
if (all.Result.Complete) return all.Result.Samples;
|
||
// Only reachable on a strict gateway (#691.3): the ceiling re-read delivered the whole ceiling without reaching
|
||
// the window end — the window may hold more, and "all values" is never answered from a truncated read.
|
||
throw new HistoryReadLimitExceededException(cap,
|
||
$"The raw read of '{fullReference}' holds more than the HistorianGateway's row ceiling ({all.Budget}, its " +
|
||
"Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); set NumValuesPerNode to page, " +
|
||
"or narrow the window.")
|
||
{ PreWindowRows = Crowding(all.Result.PreWindowRows) };
|
||
}
|
||
|
||
var asked = (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance);
|
||
var pass = await PassAsync(asked).ConfigureAwait(false);
|
||
if (pass.Result.Complete) return pass.Result.Samples;
|
||
|
||
// A strict gateway's ceiling already bounded that read (#691.3): a larger budget would only be refused again. The
|
||
// refusal carries the pre-window rows it saw (#694): the server re-reads the page that many values smaller.
|
||
if (pass.Budget < asked)
|
||
throw new HistoryReadLimitExceededException(cap,
|
||
$"The raw read of '{fullReference}' could not be read completely within the HistorianGateway's row ceiling " +
|
||
$"({pass.Budget}, its Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); lower " +
|
||
"NumValuesPerNode or narrow the window.")
|
||
{ PreWindowRows = Crowding(pass.Result.PreWindowRows) };
|
||
|
||
// The margin crowded the window out: re-read with room for the pre-window rows. A read that was itself cut short saw
|
||
// only some of them (#694 F4: a tie cluster of hundreds just before a resume point used to fail even with no
|
||
// ceiling), so the room at least doubles per re-read — bounded by MaxCrowdedReReads, and never past a known ceiling.
|
||
// #694 R2: the room is capped at the tie-cluster limit (MaxTieClusterOverfetch) — a pathological single-millisecond
|
||
// cluster on a streaming gateway would otherwise be streamed several times over for one page.
|
||
var preWindow = pass.Result.PreWindowRows;
|
||
var room = Math.Min(Math.Max(1, preWindow), _maxPreWindowRows);
|
||
for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
|
||
{
|
||
var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance);
|
||
// #695: doubling can leap over the ceiling while the cluster still fits. Spend the remaining room once.
|
||
if (_ceiling.Learned is var known and > 0 && budget > known)
|
||
{
|
||
var available = known - cap - RawAggregateReader.PreWindowAllowance;
|
||
if (available <= 0) break;
|
||
room = Math.Min(room, available);
|
||
budget = cap + room + RawAggregateReader.PreWindowAllowance;
|
||
}
|
||
if (budget <= pass.Budget) break; // no more room, never retry the same full read
|
||
pass = await PassAsync(budget).ConfigureAwait(false);
|
||
if (pass.Result.Complete) return pass.Result.Samples;
|
||
preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
|
||
if (pass.Budget < budget) break; // bounded by a ceiling learned meanwhile
|
||
if (room >= _maxPreWindowRows) break; // the limit's worth of room did not suffice
|
||
room = (int)Math.Min(_maxPreWindowRows, Math.Max(preWindow, 2L * room));
|
||
}
|
||
|
||
throw new HistoryReadLimitExceededException(cap,
|
||
$"The raw read of '{fullReference}' could not be read completely: more than {pass.Result.PreWindowRows} rows lie in " +
|
||
$"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget.")
|
||
{ PreWindowRows = preWindow };
|
||
|
||
// Pre-window rows the allowance did not cover are what crowded the read; fewer are not the cause (#694).
|
||
static int Crowding(int preWindowRows) => preWindowRows > RawAggregateReader.PreWindowAllowance ? preWindowRows : 0;
|
||
|
||
// One drained pass; a strict gateway's refusal of the budget is retried once at the ceiling it names (#691.3).
|
||
Task<((List<HistorianSample> Samples, int PreWindowRows, bool Complete) Result, int Budget)> PassAsync(int budget) =>
|
||
_ceiling.ReadAsync(budget,
|
||
b => ReadRawPassAsync(fullReference, startUtc, endUtc, cap, b, cancellationToken),
|
||
(r, _) => !r.Complete);
|
||
}
|
||
|
||
/// <summary>One forward raw read of a window, drained, with its pre-window row count and whether it is complete.</summary>
|
||
private async Task<(List<HistorianSample> Samples, int PreWindowRows, bool Complete)> ReadRawPassAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, int cap, int budget, CancellationToken cancellationToken)
|
||
{
|
||
var samples = new List<HistorianSample>();
|
||
long received = 0;
|
||
var preWindow = 0;
|
||
var reachedEnd = false;
|
||
await foreach (var sample in _client
|
||
.ReadRawAsync(fullReference, RawAggregateReader.RequestStartFor(startUtc), endUtc, budget, cancellationToken)
|
||
.ConfigureAwait(false))
|
||
{
|
||
received++;
|
||
// Filter while draining — the rest of the (bounded) stream is consumed, never abandoned.
|
||
if (RawAggregateReader.IsWindowSample(sample, startUtc, endUtc))
|
||
{
|
||
if (cap == 0 || samples.Count < cap) samples.Add(sample);
|
||
}
|
||
// A row before the window — or, for a window starting at/before the epoch (no look-back), AVEVA's synthetic row at
|
||
// the query start, which is no sample either (#694 F1: it used to set reachedEnd, so a ceiling-truncated read
|
||
// passed as complete) — is a pre-window row.
|
||
else if (sample.Timestamp is { } ts
|
||
&& (ts.ToDateTime() < startUtc || !RawAggregateReader.IsAfterFloor(ts.ToDateTime()))) preWindow++;
|
||
else if (sample.Timestamp is not null) reachedEnd = true;
|
||
}
|
||
|
||
// cap == 0 ("all values") is complete only when the stream proves it: short of the budget (always, unless a strict
|
||
// gateway's ceiling became the budget — #691.3) or past the window end.
|
||
var complete = received < budget || reachedEnd || (cap > 0 && samples.Count >= cap);
|
||
return (samples, preWindow, complete);
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>Per-sample aggregates (#681)</b> — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
|
||
/// Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
|
||
/// (<see cref="RawAggregateReader"/>): identical values, status, Calculated/Partial bits and timestamps to
|
||
/// SqlServer/QuestDB for the same samples, forward and reverse (#677). The in-window raw rows are capped by
|
||
/// <c>ServerHistorian:MaxTimeWeightedRawRows</c> (<see cref="HistoryReadLimitExceededException"/>, not counted as
|
||
/// a backend failure).
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
|
||
/// the interval start and the Calculated bit; AVEVA's extra row for the interval before the window is dropped and
|
||
/// an Uncertain interval keeps its value (#685). Both directions use the SQL backends' interval grid with forward
|
||
/// requests only — a reverse read (which the gateway refuses) anchored at its later time (#677), the clipped interval
|
||
/// read over exactly its span and marked Partial (#691.2, <see cref="NativeAggregatePlan"/>). Everything else is <see cref="HistoryAggregateNotSupportedException"/>
|
||
/// (see <see cref="ServedAggregates"/>).
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <exception cref="ArgumentOutOfRangeException">A raw-evaluated read whose window holds more than
|
||
/// <see cref="BucketPlan.MaxBuckets"/> intervals.</exception>
|
||
public async Task<HistoryReadResult> ReadProcessedAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||
{
|
||
if (RawAggregateReader.Serves(aggregate))
|
||
return await ReadProcessedFromRawAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
|
||
// Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
|
||
// request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
|
||
var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
|
||
// Planned OUTSIDE the failure-recording try: too many intervals is a client request limit, not a backend fault.
|
||
return await ReadNativeAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
}
|
||
|
||
/// <summary>
|
||
/// A native TimeAverage / Total read on the SQL backends' interval grid, either direction (#691.2): forward native
|
||
/// requests only, the clipped interval read over exactly its span. See <see cref="NativeAggregatePlan"/>.
|
||
/// </summary>
|
||
private async Task<HistoryReadResult> ReadNativeAsync(string fullReference, BucketPlan plan, RetrievalMode mode,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
|
||
try
|
||
{
|
||
if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, _time.GetUtcNow().UtcDateTime)), ContinuationPoint: null);
|
||
}
|
||
var byBucket = new Dictionary<long, DataValueSnapshot>();
|
||
foreach (var request in NativeAggregatePlan.Requests(plan))
|
||
{
|
||
var buckets = new List<HistorianAggregateSample>();
|
||
await foreach (var bucket in _client
|
||
.ReadAggregateAsync(fullReference, request.StartUtc, request.EndUtc, mode, request.Interval, cancellationToken)
|
||
.ConfigureAwait(false))
|
||
{
|
||
buckets.Add(bucket); // drained to completion, mapped afterwards
|
||
}
|
||
NativeAggregatePlan.Place(plan, request, buckets, byBucket);
|
||
}
|
||
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(NativeAggregatePlan.Assemble(plan, byBucket), ContinuationPoint: null);
|
||
}
|
||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||
{
|
||
throw; // caller-requested, not a backend fault
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>Serves a per-sample aggregate from a raw read (#681); see <see cref="RawAggregateReader"/>.</summary>
|
||
private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||
{
|
||
// Planned OUTSIDE the failure-recording try (as on the SQL backends): too many intervals is a client request
|
||
// limit, not a backend fault.
|
||
var plan = BucketPlan.Create(startUtc, endUtc, interval);
|
||
try
|
||
{
|
||
var now = _time.GetUtcNow().UtcDateTime;
|
||
if (plan.BucketCount > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, now)), ContinuationPoint: null);
|
||
}
|
||
var values = await RawAggregateReader
|
||
.ReadAsync(_client, _ceiling, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(values, ContinuationPoint: null);
|
||
}
|
||
catch (HistoryReadLimitExceededException)
|
||
{
|
||
// A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
|
||
throw;
|
||
}
|
||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||
{
|
||
// The caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault.
|
||
throw;
|
||
}
|
||
catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
|
||
{
|
||
// The gateway refused the row budget (Historian:ReadLimits:MaxValuesCeiling with
|
||
// AllowStreamingBeyondCeiling=false): the same client-request limit as the cap, not a backend fault.
|
||
throw new HistoryReadLimitExceededException(_maxRawRows,
|
||
$"The HistorianGateway refused the raw row budget of the processed read of '{fullReference}' (its " +
|
||
"Historian:ReadLimits:MaxValuesCeiling); lower ServerHistorian:MaxTimeWeightedRawRows below that ceiling " +
|
||
"or narrow the window.");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// True when <paramref name="ex"/> (or an inner exception — the client package wraps the <see cref="RpcException"/>)
|
||
/// is the gateway's <c>InvalidArgument</c> "max_values exceeds the server ceiling" rejection.
|
||
/// </summary>
|
||
internal static bool IsMaxValuesCeilingRejection(Exception ex) => GatewayReadLimits.IsCeilingRejection(ex);
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// #693: served from RAW gateway reads — each requested time's stored sample at-or-before and after it — evaluated with
|
||
/// the SQL backends' shared <see cref="AtTimeEvaluator"/> (Part 11 §6.5.4), so the Gateway answers what SqlServer and
|
||
/// QuestDB answer for the same stored samples; see <see cref="GatewayAtTimeReader"/> for the read plan and its cost.
|
||
/// AVEVA's own interpolated answer (the gateway's <c>ReadAtTime</c>) returned the previous value as Good at a stored
|
||
/// NULL's time, null + Bad at a Good sample right after a NULL, a drifted value at an exact stored sample, and no
|
||
/// Interpolated bit; it is only used when <c>ServerHistorian:GatewayAtTimeFromRaw = false</c> (kill switch).
|
||
/// A gateway value is a Double (or a String) whatever the node's type (AVEVA stores a Boolean as a 0/1 <c>Int2</c>), so
|
||
/// without a type hint every numeric tag is interpolated linearly; the node manager passes the node's type through
|
||
/// <see cref="ReadAtTimeAsync(string, IReadOnlyList{DateTime}, TypeCode?, CancellationToken)"/>.
|
||
/// </remarks>
|
||
public Task<HistoryReadResult> ReadAtTimeAsync(
|
||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
||
ReadAtTimeAsync(fullReference, timestampsUtc, valueType: null, cancellationToken);
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// #693: the hint makes the AtTime rules type-exact, as on the SQL backends — the gateway's Double rows are converted to
|
||
/// the node's type BEFORE the shared evaluator runs (Boolean = value ≠ 0 ⇒ stepped; an integer type ⇒ Int64 ⇒
|
||
/// interpolated and rounded half away from zero, then returned as the node's integer type; String ⇒ stepped; Float /
|
||
/// Double / no hint ⇒ linear Double). Ignored with the kill switch (AVEVA's own answer).
|
||
/// </remarks>
|
||
public async Task<HistoryReadResult> ReadAtTimeAsync(
|
||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, TypeCode? valueType, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(timestampsUtc);
|
||
if (!_atTimeFromRaw) return await ReadAtTimeNativeAsync(fullReference, timestampsUtc, cancellationToken).ConfigureAwait(false);
|
||
try
|
||
{
|
||
var now = _time.GetUtcNow().UtcDateTime;
|
||
if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
RecordOutcome(success: true, error: null);
|
||
return MissingAtTime(timestampsUtc, now);
|
||
}
|
||
var values = await GatewayAtTimeReader
|
||
.ReadAsync(_client, _ceiling, fullReference, timestampsUtc, now, cancellationToken, valueType)
|
||
.ConfigureAwait(false);
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(values, ContinuationPoint: null);
|
||
}
|
||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||
{
|
||
throw; // caller-requested, not a backend fault
|
||
}
|
||
catch (HistoryReadLimitExceededException)
|
||
{
|
||
throw; // a client-request limit (a pathological tie cluster), not a backend fault
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>The pre-#693 AtTime: AVEVA's own interpolation, aligned to the request (kill switch only).</summary>
|
||
private async Task<HistoryReadResult> ReadAtTimeNativeAsync(
|
||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||
{
|
||
try
|
||
{
|
||
if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
|
||
{
|
||
RecordOutcome(success: true, error: null);
|
||
return MissingAtTime(timestampsUtc, _time.GetUtcNow().UtcDateTime);
|
||
}
|
||
|
||
var samples = await _client
|
||
.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
var aligned = AlignAtTimeSnapshots(timestampsUtc, samples);
|
||
RecordOutcome(success: true, error: null);
|
||
return new HistoryReadResult(aligned, ContinuationPoint: null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// <c>false</c> (Gitea #686): the gateway's <c>ReadRaw</c> streams ascending only and rejects <c>end < start</c>
|
||
/// (<c>InvalidArgument</c>). The server serves reverse and end-only capped raw reads from bounded forward reads
|
||
/// instead, and never sends this source a reverse window (<see cref="ReadRawAsync"/> refuses one unsent).
|
||
/// </remarks>
|
||
public bool SupportsReverseRaw => false;
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// Gitea #694: <c>0</c> until a strict gateway (<c>AllowStreamingBeyondCeiling=false</c>) has refused a budget and
|
||
/// named its ceiling (#691.3, <see cref="GatewayRowCeiling"/>); then that ceiling less the
|
||
/// <see cref="RawAggregateReader.PreWindowAllowance"/> a capped read adds for AVEVA's boundary row — so a page of this
|
||
/// size asks for exactly the ceiling. Back to <c>0</c> when a re-probe finds the gateway streaming again.
|
||
/// </remarks>
|
||
public int MaxRawValuesPerRead => _ceiling.Learned is var c and > 0
|
||
? Math.Max(1, c - RawAggregateReader.PreWindowAllowance)
|
||
: 0;
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// <c>false</c>: the gateway's <c>ReadEvents</c> streams ascending only (Gitea #675). The server never sends it a
|
||
/// reverse read; since #691.4 it walks one back from bounded forward reads (ties sorted by EventId — the gateway
|
||
/// orders by event time only).
|
||
/// </remarks>
|
||
public bool SupportsReverseEvents => false;
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks><see cref="ServedAggregates"/>; every other member answers <c>BadAggregateNotSupported</c> and is not advertised.</remarks>
|
||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => ServedAggregates;
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// A reverse (newest-first) read is refused with <see cref="NotSupportedException"/> (Gitea #675) — the gateway's
|
||
/// <c>ReadEvents</c> streams ascending only and rejects <c>end < start</c>. The server never calls this with
|
||
/// <c>reverse</c>: it walks a reverse read back from bounded forward reads (#691.4). Nothing is sent.
|
||
/// </remarks>
|
||
public Task<HistoricalEventsResult> ReadEventsAsync(
|
||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse,
|
||
CancellationToken cancellationToken) =>
|
||
reverse
|
||
? throw new NotSupportedException(
|
||
"The HistorianGateway backend reads events oldest-first only; a reverse (newest-first) events read is not supported.")
|
||
: ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
|
||
|
||
/// <inheritdoc />
|
||
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
try
|
||
{
|
||
var hasCap = maxEvents > 0;
|
||
// Bounded initial capacity: an uncapped read arrives as int.MaxValue ("Array dimensions exceeded", #691.4 review).
|
||
var collected = new List<HistoricalEvent>(hasCap ? Math.Min(maxEvents, 1024) : 0);
|
||
var truncated = false;
|
||
|
||
await foreach (var wireEvent in _client
|
||
.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken)
|
||
.ConfigureAwait(false))
|
||
{
|
||
var mapped = EventMapper.ToHistoricalEvent(wireEvent);
|
||
|
||
// Defensive client-side source filter: the gateway's SQL ReadEvents source filter
|
||
// may not be present, so drop any event whose source does not match the request.
|
||
if (sourceName is not null && !string.Equals(mapped.SourceName, sourceName, StringComparison.Ordinal))
|
||
{
|
||
continue;
|
||
}
|
||
|
||
// One more matching event arriving once the cap is full means the result is
|
||
// truncated — stop draining and flag it.
|
||
if (hasCap && collected.Count == maxEvents)
|
||
{
|
||
truncated = true;
|
||
break;
|
||
}
|
||
|
||
collected.Add(mapped);
|
||
}
|
||
|
||
RecordOutcome(success: true, error: null);
|
||
// A non-null, opaque token signals truncation to the caller.
|
||
// The gateway has no resumable cursor, so the token's contents carry no paging state —
|
||
// its presence alone is the "more events exist" signal. A fresh array per call keeps it
|
||
// from being shared/mutated.
|
||
return new HistoricalEventsResult(collected, truncated ? new byte[] { 0x01 } : null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
RecordReadFailure(ex);
|
||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
public HistorianHealthSnapshot GetHealthSnapshot()
|
||
{
|
||
lock (_healthLock)
|
||
{
|
||
return new HistorianHealthSnapshot(
|
||
TotalQueries: _totalQueries,
|
||
TotalSuccesses: _totalSuccesses,
|
||
TotalFailures: _totalFailures,
|
||
ConsecutiveFailures: _consecutiveFailures,
|
||
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,
|
||
ActiveProcessNode: null,
|
||
ActiveEventNode: null,
|
||
Nodes: []);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Reconciles a gateway at-time reply against the requested timestamps to honour the
|
||
/// <see cref="IHistorianDataSource.ReadAtTimeAsync(string, IReadOnlyList{DateTime}, CancellationToken)"/> contract: exactly one snapshot per
|
||
/// requested timestamp, in request order. Returned samples are indexed by timestamp ticks;
|
||
/// any requested timestamp the gateway did not return is filled with a Bad-quality
|
||
/// (<c>0x80000000</c>) snapshot stamped at the requested time rather than positionally
|
||
/// misaligning values. The alignment logic was ported from the now-retired Wonderware
|
||
/// client's at-time snapshot reconciliation.
|
||
/// </summary>
|
||
private static IReadOnlyList<DataValueSnapshot> AlignAtTimeSnapshots(
|
||
IReadOnlyList<DateTime> timestampsUtc, IReadOnlyList<HistorianSample> samples)
|
||
{
|
||
// Index returned samples by timestamp ticks. Duplicate timestamps keep the first.
|
||
var byTicks = new Dictionary<long, HistorianSample>(samples.Count);
|
||
foreach (var sample in samples)
|
||
{
|
||
if (sample.Timestamp is null) continue;
|
||
byTicks.TryAdd(sample.Timestamp.ToDateTime().Ticks, sample);
|
||
}
|
||
|
||
var result = new DataValueSnapshot[timestampsUtc.Count];
|
||
for (var i = 0; i < timestampsUtc.Count; i++)
|
||
{
|
||
var requested = DateTime.SpecifyKind(timestampsUtc[i], DateTimeKind.Utc);
|
||
if (byTicks.TryGetValue(requested.Ticks, out var sample))
|
||
{
|
||
// Reuse the shared sample mapper for value + quality, then re-stamp the source
|
||
// timestamp to the requested time per the ReadAtTime contract.
|
||
result[i] = SampleMapper.ToSnapshot(sample) with { SourceTimestampUtc = requested };
|
||
}
|
||
else
|
||
{
|
||
// Gap — gateway returned no sample for this timestamp. Per the contract this is a
|
||
// Bad-quality snapshot stamped at the requested time, not a dropped row.
|
||
result[i] = new DataValueSnapshot(
|
||
Value: null,
|
||
StatusCode: 0x80000000u, // Bad
|
||
SourceTimestampUtc: requested,
|
||
ServerTimestampUtc: DateTime.UtcNow);
|
||
}
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
/// <summary>No-data values for a missing tag, stamped with the client's requested times.</summary>
|
||
private static HistoryReadResult MissingAtTime(IReadOnlyList<DateTime> times, DateTime now) =>
|
||
new(times.Select(t => AtTimeEvaluator.Evaluate(t, null, null, now, now)).ToArray(), ContinuationPoint: null);
|
||
|
||
/// <summary>Classifies actual gRPC/package transport failures without treating refusals as outages (#697).</summary>
|
||
private static bool IsTransportFailure(Exception ex) => ex switch
|
||
{
|
||
ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException => true,
|
||
RpcException rpc => rpc.StatusCode is Grpc.Core.StatusCode.Unavailable or Grpc.Core.StatusCode.DeadlineExceeded,
|
||
TimeoutException => true,
|
||
_ => ex.InnerException is { } inner && IsTransportFailure(inner),
|
||
};
|
||
|
||
/// <summary>
|
||
/// Records a failed read: bumps the health counters and logs a generic, redaction-safe
|
||
/// debug line (no tag, host, or value). The exception itself is rethrown by the caller.
|
||
/// </summary>
|
||
private void RecordReadFailure(Exception ex)
|
||
{
|
||
RecordOutcome(success: false, error: ex.Message);
|
||
_logger.LogDebug("Historian gateway read operation failed and was recorded as a health failure.");
|
||
}
|
||
|
||
/// <summary>
|
||
/// Records the outcome of a single read — increments <c>_totalQueries</c> and exactly one of
|
||
/// <c>_totalSuccesses</c> / <c>_totalFailures</c> under a single <c>_healthLock</c>
|
||
/// acquisition so a concurrent <see cref="GetHealthSnapshot"/> never observes a torn state.
|
||
/// </summary>
|
||
private void RecordOutcome(bool success, string? error)
|
||
{
|
||
lock (_healthLock)
|
||
{
|
||
_totalQueries++;
|
||
if (success)
|
||
{
|
||
_totalSuccesses++;
|
||
_consecutiveFailures = 0;
|
||
_lastSuccessUtc = DateTime.UtcNow;
|
||
}
|
||
else
|
||
{
|
||
_totalFailures++;
|
||
_consecutiveFailures++;
|
||
_lastFailureUtc = DateTime.UtcNow;
|
||
_lastError = error;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>Disposes the underlying gateway client. Prefer <see cref="DisposeAsync"/>.</summary>
|
||
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
|
||
|
||
/// <summary>Asynchronously disposes the underlying gateway client.</summary>
|
||
/// <returns>A task that completes when the client has been disposed.</returns>
|
||
public ValueTask DisposeAsync() => _client.DisposeAsync();
|
||
}
|