Joseph Doherty
383 lines
16 KiB
C#
383 lines
16 KiB
C#
using System.Runtime.CompilerServices;
|
|
using Google.Protobuf;
|
|
using Grpc.Core;
|
|
using AVEVA.Historian.Client.Models;
|
|
using AVEVA.Historian.Client.Wcf;
|
|
using GrpcRetrieval = ArchestrA.Grpc.Contract.Retrieval;
|
|
|
|
namespace AVEVA.Historian.Client.Grpc;
|
|
|
|
/// <summary>
|
|
/// 2023 R2 gRPC read orchestrator. Mirrors <see cref="HistorianWcfReadOrchestrator"/> over the
|
|
/// gRPC transport: the same native binary buffers travel inside protobuf <c>bytes</c> fields,
|
|
/// and the same serializers/parsers (<see cref="HistorianNativeHandshake"/>,
|
|
/// <see cref="HistorianDataQueryProtocol"/>) are reused unchanged.
|
|
///
|
|
/// Operation mapping (2020 WCF → 2023 R2 gRPC):
|
|
/// Hist.GetInterfaceVersion → HistoryService.GetInterfaceVersion
|
|
/// Hist.ValidateClientCredential (loop) → StorageService.ValidateClientCredential (loop)
|
|
/// Hist.OpenConnection2 → HistoryService.OpenConnection
|
|
/// Retr.StartQuery2 → RetrievalService.StartQuery
|
|
/// Retr.GetNextQueryResultBuffer2 (loop) → RetrievalService.GetNextQueryResultBuffer (loop)
|
|
/// Retr.EndQuery2 → RetrievalService.EndQuery
|
|
///
|
|
/// LIVE-VERIFIED 2026-06-21 against a real 2023 R2 server (interface versions: History=12,
|
|
/// Retrieval=4, Storage=4). The SSPI/Negotiate token loop maps to
|
|
/// <c>StorageService.ValidateClientCredential(Handle, InBuff)→(status, OutBuff)</c> — the op that
|
|
/// kept the 2020 inBuff/outBuff token framing. The gRPC HistoryService dropped
|
|
/// ValidateClientCredential and gained <c>ExchangeKey</c>, but ExchangeKey is a SEPARATE
|
|
/// key-exchange/cert-path op, NOT the Negotiate loop: feeding it an NTLM token is rejected at
|
|
/// round 0 regardless of credentials. An earlier revision wrongly routed the loop to ExchangeKey;
|
|
/// routing it to StorageService.ValidateClientCredential completes the full read chain. The byte
|
|
/// payloads (OpenConnection3 v6, token framing, DataQueryRequest, row buffers) are the proven 2020
|
|
/// protocol and transfer unchanged — only the History interface integer differs (12 vs the 2020
|
|
/// value 11), and that version is buffer-compatible (a live read returns rows).
|
|
/// </summary>
|
|
internal sealed class HistorianGrpcReadOrchestrator
|
|
{
|
|
private const ushort StartQueryRequestType = HistorianDataQueryProtocol.QueryRequestTypeData;
|
|
|
|
private readonly HistorianClientOptions _options;
|
|
|
|
public HistorianGrpcReadOrchestrator(HistorianClientOptions options)
|
|
{
|
|
_options = options ?? throw new ArgumentNullException(nameof(options));
|
|
}
|
|
|
|
public async IAsyncEnumerable<HistorianSample> ReadRawAsync(
|
|
string tag,
|
|
DateTime startUtc,
|
|
DateTime endUtc,
|
|
int maxValues,
|
|
[EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
ValidateAuth();
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
|
|
IReadOnlyList<HistorianSample> rows = await Task.Run(
|
|
() => RunRawChain(tag, startUtc, endUtc, maxValues, cancellationToken), cancellationToken).ConfigureAwait(false);
|
|
foreach (HistorianSample sample in rows)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
yield return sample;
|
|
}
|
|
}
|
|
|
|
public async IAsyncEnumerable<HistorianAggregateSample> ReadAggregateAsync(
|
|
string tag,
|
|
DateTime startUtc,
|
|
DateTime endUtc,
|
|
RetrievalMode mode,
|
|
TimeSpan interval,
|
|
[EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
ValidateAuth();
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
|
|
IReadOnlyList<HistorianAggregateSample> rows = await Task.Run(
|
|
() => RunAggregateChain(tag, startUtc, endUtc, mode, interval, cancellationToken), cancellationToken).ConfigureAwait(false);
|
|
foreach (HistorianAggregateSample sample in rows)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
yield return sample;
|
|
}
|
|
}
|
|
|
|
public Task<IReadOnlyList<HistorianSample>> ReadAtTimeAsync(
|
|
string tag,
|
|
IReadOnlyList<DateTime> timestampsUtc,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
ValidateAuth();
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
return Task.Run<IReadOnlyList<HistorianSample>>(() => RunAtTimeChain(tag, timestampsUtc, cancellationToken), cancellationToken);
|
|
}
|
|
|
|
private void ValidateAuth()
|
|
{
|
|
if (!_options.IntegratedSecurity && string.IsNullOrEmpty(_options.UserName))
|
|
{
|
|
throw new ProtocolEvidenceMissingException(
|
|
"Managed gRPC read flow currently requires IntegratedSecurity or an explicit UserName + Password.");
|
|
}
|
|
}
|
|
|
|
// Spike/Phase-1 seam (pending.md A1): run a raw query against an EXTERNALLY-supplied, already-
|
|
// authenticated connection + client handle — i.e. NO Create()/handshake here. RunRawChain delegates
|
|
// to this so the per-call path and the reuse path share one query implementation (DRY). The handshake
|
|
// reuse-probe test drives this directly to measure whether the server honors a reused session.
|
|
internal List<HistorianSample> RunRawQueryOnSession(
|
|
HistorianGrpcConnection connection,
|
|
uint clientHandle,
|
|
string tag,
|
|
DateTime startUtc,
|
|
DateTime endUtc,
|
|
int maxValues,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
HistorianDataQueryRequest request = HistorianWcfReadOrchestrator.BuildDataQueryRequest(tag, startUtc, endUtc, maxValues);
|
|
return RunQuery(connection, clientHandle, request, maxValues, cancellationToken);
|
|
}
|
|
|
|
private List<HistorianSample> RunRawChain(string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken cancellationToken)
|
|
{
|
|
using HistorianGrpcConnection connection = HistorianGrpcChannelFactory.Create(_options);
|
|
uint clientHandle = OpenAuthenticatedConnection(connection, cancellationToken);
|
|
return RunRawQueryOnSession(connection, clientHandle, tag, startUtc, endUtc, maxValues, cancellationToken);
|
|
}
|
|
|
|
// Spike/Phase-1 seam (pending.md A1): run an aggregate query against an EXTERNALLY-supplied,
|
|
// already-authenticated connection + client handle — i.e. NO Create()/handshake here.
|
|
// RunAggregateChain delegates to this so the per-call path and the reuse path share one query
|
|
// implementation (DRY).
|
|
internal List<HistorianAggregateSample> RunAggregateQueryOnSession(
|
|
HistorianGrpcConnection connection,
|
|
uint clientHandle,
|
|
string tag,
|
|
DateTime startUtc,
|
|
DateTime endUtc,
|
|
RetrievalMode mode,
|
|
TimeSpan interval,
|
|
CancellationToken ct)
|
|
{
|
|
return RunAggregateQuery(connection, clientHandle, tag, startUtc, endUtc, mode, interval, ct);
|
|
}
|
|
|
|
private List<HistorianAggregateSample> RunAggregateChain(
|
|
string tag, DateTime startUtc, DateTime endUtc, RetrievalMode mode, TimeSpan interval, CancellationToken cancellationToken)
|
|
{
|
|
using HistorianGrpcConnection connection = HistorianGrpcChannelFactory.Create(_options);
|
|
uint clientHandle = OpenAuthenticatedConnection(connection, cancellationToken);
|
|
return RunAggregateQueryOnSession(connection, clientHandle, tag, startUtc, endUtc, mode, interval, cancellationToken);
|
|
}
|
|
|
|
// Spike/Phase-1 seam (pending.md A1): run an at-time query against an EXTERNALLY-supplied,
|
|
// already-authenticated connection + client handle — i.e. NO Create()/handshake here.
|
|
// RunAtTimeChain delegates to this so the per-call path and the reuse path share one
|
|
// implementation (DRY).
|
|
internal List<HistorianSample> RunAtTimeOnSession(
|
|
HistorianGrpcConnection connection,
|
|
uint clientHandle,
|
|
string tag,
|
|
IReadOnlyList<DateTime> timestampsUtc,
|
|
CancellationToken ct)
|
|
{
|
|
if (timestampsUtc.Count == 0)
|
|
{
|
|
return [];
|
|
}
|
|
|
|
List<HistorianSample> results = new(timestampsUtc.Count);
|
|
foreach (DateTime ts in timestampsUtc)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
DateTime tsUtc = ts.ToUniversalTime();
|
|
List<HistorianAggregateSample> aggregates = RunAggregateQuery(
|
|
connection,
|
|
clientHandle,
|
|
tag,
|
|
tsUtc - TimeSpan.FromTicks(1),
|
|
tsUtc + TimeSpan.FromTicks(1),
|
|
RetrievalMode.Interpolated,
|
|
TimeSpan.FromTicks(2),
|
|
ct);
|
|
|
|
if (aggregates.Count == 0)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
HistorianAggregateSample chosen = aggregates[0];
|
|
results.Add(new HistorianSample(
|
|
TagName: chosen.TagName,
|
|
TimestampUtc: tsUtc,
|
|
NumericValue: chosen.Value,
|
|
StringValue: null,
|
|
Quality: chosen.Quality,
|
|
QualityDetail: chosen.QualityDetail,
|
|
OpcQuality: chosen.OpcQuality,
|
|
PercentGood: 100));
|
|
}
|
|
|
|
return results;
|
|
}
|
|
|
|
private List<HistorianSample> RunAtTimeChain(string tag, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
|
{
|
|
if (timestampsUtc.Count == 0)
|
|
{
|
|
return [];
|
|
}
|
|
|
|
using HistorianGrpcConnection connection = HistorianGrpcChannelFactory.Create(_options);
|
|
uint clientHandle = OpenAuthenticatedConnection(connection, cancellationToken);
|
|
return RunAtTimeOnSession(connection, clientHandle, tag, timestampsUtc, cancellationToken);
|
|
}
|
|
|
|
private uint OpenAuthenticatedConnection(HistorianGrpcConnection connection, CancellationToken cancellationToken)
|
|
=> HistorianGrpcHandshake.OpenAuthenticatedConnection(connection, _options, cancellationToken);
|
|
|
|
private List<HistorianSample> RunQuery(
|
|
HistorianGrpcConnection connection,
|
|
uint clientHandle,
|
|
HistorianDataQueryRequest request,
|
|
int maxValues,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var retrievalClient = new GrpcRetrieval.RetrievalService.RetrievalServiceClient(connection.Channel);
|
|
GrpcRetrieval.GetRetrievalInterfaceVersionResponse retrievalVersion = retrievalClient.GetRetrievalInterfaceVersion(
|
|
new GrpcRetrieval.GetRetrievalInterfaceVersionRequest(), null, Deadline(), cancellationToken);
|
|
HistorianServerVersionGate.Validate(HistorianServiceInterface.Retrieval, retrievalVersion.UiVersion, _options);
|
|
|
|
byte[] requestBuffer = HistorianDataQueryProtocol.SerializeFullHistoryRequest(request);
|
|
uint queryHandle = StartQuery(retrievalClient, clientHandle, requestBuffer, "raw", cancellationToken);
|
|
|
|
try
|
|
{
|
|
List<HistorianSample> samples = [];
|
|
while (true)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
(byte[] resultBuffer, byte[] errorBuffer) = GetNextResultBuffer(retrievalClient, clientHandle, queryHandle, "raw", cancellationToken);
|
|
|
|
if (!HistorianDataQueryProtocol.TryParseGetNextQueryResultBufferRows(resultBuffer, errorBuffer, out IReadOnlyList<HistorianSample> rows, out bool hasMoreData))
|
|
{
|
|
throw new InvalidOperationException($"gRPC GetNextQueryResultBuffer returned an unparsable result buffer (length={resultBuffer.Length}).");
|
|
}
|
|
|
|
foreach (HistorianSample sample in rows)
|
|
{
|
|
samples.Add(sample);
|
|
if (samples.Count >= maxValues)
|
|
{
|
|
return samples;
|
|
}
|
|
}
|
|
|
|
if (!hasMoreData)
|
|
{
|
|
return samples;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
EndQuerySafely(retrievalClient, clientHandle, queryHandle);
|
|
}
|
|
}
|
|
|
|
private List<HistorianAggregateSample> RunAggregateQuery(
|
|
HistorianGrpcConnection connection,
|
|
uint clientHandle,
|
|
string tag,
|
|
DateTime startUtc,
|
|
DateTime endUtc,
|
|
RetrievalMode mode,
|
|
TimeSpan interval,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var retrievalClient = new GrpcRetrieval.RetrievalService.RetrievalServiceClient(connection.Channel);
|
|
GrpcRetrieval.GetRetrievalInterfaceVersionResponse retrievalVersion = retrievalClient.GetRetrievalInterfaceVersion(
|
|
new GrpcRetrieval.GetRetrievalInterfaceVersionRequest(), null, Deadline(), cancellationToken);
|
|
HistorianServerVersionGate.Validate(HistorianServiceInterface.Retrieval, retrievalVersion.UiVersion, _options);
|
|
|
|
HistorianDataQueryRequest request = HistorianWcfReadOrchestrator.BuildAggregateQueryRequest(tag, startUtc, endUtc, mode, interval);
|
|
byte[] requestBuffer = HistorianDataQueryProtocol.SerializeFullHistoryRequest(request);
|
|
uint queryHandle = StartQuery(retrievalClient, clientHandle, requestBuffer, $"aggregate {mode}", cancellationToken);
|
|
|
|
try
|
|
{
|
|
List<HistorianAggregateSample> samples = [];
|
|
while (true)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
(byte[] resultBuffer, byte[] errorBuffer) = GetNextResultBuffer(retrievalClient, clientHandle, queryHandle, $"aggregate {mode}", cancellationToken);
|
|
|
|
if (!HistorianDataQueryProtocol.TryParseGetNextQueryResultBufferAggregateRows(
|
|
resultBuffer, errorBuffer, mode, interval, out IReadOnlyList<HistorianAggregateSample> rows, out bool hasMoreData))
|
|
{
|
|
throw new InvalidOperationException($"gRPC GetNextQueryResultBuffer (aggregate {mode}) returned an unparsable buffer (length={resultBuffer.Length}).");
|
|
}
|
|
|
|
samples.AddRange(rows);
|
|
if (!hasMoreData)
|
|
{
|
|
return samples;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
EndQuerySafely(retrievalClient, clientHandle, queryHandle);
|
|
}
|
|
}
|
|
|
|
private uint StartQuery(
|
|
GrpcRetrieval.RetrievalService.RetrievalServiceClient client,
|
|
uint clientHandle,
|
|
byte[] requestBuffer,
|
|
string label,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
GrpcRetrieval.StartQueryResponse response = client.StartQuery(
|
|
new GrpcRetrieval.StartQueryRequest
|
|
{
|
|
UiHandle = clientHandle,
|
|
UiQueryRequestType = StartQueryRequestType,
|
|
BtRequestBuffer = ByteString.CopyFrom(requestBuffer)
|
|
},
|
|
null,
|
|
Deadline(),
|
|
cancellationToken);
|
|
|
|
if (!(response.Status?.BSuccess ?? false))
|
|
{
|
|
byte[] err = response.Status?.BtError?.ToByteArray() ?? [];
|
|
throw new InvalidOperationException($"gRPC StartQuery ({label}) failed (errorLen={err.Length}).");
|
|
}
|
|
|
|
return response.UiQueryHandle;
|
|
}
|
|
|
|
private (byte[] ResultBuffer, byte[] ErrorBuffer) GetNextResultBuffer(
|
|
GrpcRetrieval.RetrievalService.RetrievalServiceClient client,
|
|
uint clientHandle,
|
|
uint queryHandle,
|
|
string label,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
GrpcRetrieval.GetNextQueryResultBufferResponse response = client.GetNextQueryResultBuffer(
|
|
new GrpcRetrieval.GetNextQueryResultBufferRequest { UiHandle = clientHandle, UiQueryHandle = queryHandle },
|
|
null,
|
|
Deadline(),
|
|
cancellationToken);
|
|
|
|
byte[] errorBuffer = response.Status?.BtError?.ToByteArray() ?? [];
|
|
if (!(response.Status?.BSuccess ?? false))
|
|
{
|
|
throw new InvalidOperationException($"gRPC GetNextQueryResultBuffer ({label}) failed (errorLen={errorBuffer.Length}).");
|
|
}
|
|
|
|
byte[] resultBuffer = response.BtQueryResult?.ToByteArray() ?? [];
|
|
return (resultBuffer, errorBuffer);
|
|
}
|
|
|
|
private void EndQuerySafely(GrpcRetrieval.RetrievalService.RetrievalServiceClient client, uint clientHandle, uint queryHandle)
|
|
{
|
|
try
|
|
{
|
|
client.EndQuery(
|
|
new GrpcRetrieval.EndQueryRequest { UiHandle = clientHandle, UiQueryHandle = queryHandle },
|
|
null,
|
|
Deadline(),
|
|
CancellationToken.None);
|
|
}
|
|
catch
|
|
{
|
|
// Best-effort cleanup; the read result is already collected.
|
|
}
|
|
}
|
|
|
|
private DateTime Deadline() => DateTime.UtcNow.Add(_options.RequestTimeout);
|
|
}
|