Add 2023 R2 gRPC transport (RemoteGrpc) reusing native byte payloads
Stands up HistorianTransport.RemoteGrpc end-to-end for the read path, built on the recovered 2023 R2 gRPC contract (gRPC-Web/HTTP-1.1, port 32565, gzip). The opaque protobuf `bytes` fields carry the SAME native binary payloads as the 2020 WCF/MDAS path, so the proven serializers and parsers are reused unchanged. - Grpc/Protos/*.proto: 6 protoc-validated contracts recovered from embedded FileDescriptors (authoritative, not guessed). - Grpc/HistorianGrpcChannelFactory: GrpcWebHandler/HTTP-1.1 channel, ResolvePort/ResolveAddress, optional TLS + gzip. - Grpc/HistorianGrpcReadOrchestrator: mirrors the WCF read chain over gRPC; auth uses HistoryService.ExchangeKey (the gRPC ValCl op). - Wcf/HistorianNativeHandshake: transport-agnostic Open2 request builder + SSPI/Negotiate token loop + response decode, shared by WCF and gRPC. - Op map (2020 -> gRPC): ValCl->ExchangeKey, Open2->OpenConnection, StartQuery2->StartQuery, GetNextQueryResultBuffer2->GetNextQueryResultBuffer. - HistorianClientOptions: DefaultGrpcPort=32565, GrpcUseTls. - csproj: Google.Protobuf, Grpc.Net.Client(.Web), Grpc.Tools codegen. Not yet live-verified against a 2023 R2 server: ExchangeKey is the first thing to revisit if a live server rejects the handshake; the inner byte payloads are the proven 2020 protocol. Gated live test via HISTORIAN_GRPC_HOST. 188 unit tests green; build clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,363 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using Google.Protobuf;
|
||||
using Grpc.Core;
|
||||
using AVEVA.Historian.Client.Models;
|
||||
using AVEVA.Historian.Client.Wcf;
|
||||
using GrpcHistory = ArchestrA.Grpc.Contract.History;
|
||||
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) → HistoryService.ExchangeKey (loop)
|
||||
/// Hist.OpenConnection2 → HistoryService.OpenConnection
|
||||
/// Retr.StartQuery2 → RetrievalService.StartQuery
|
||||
/// Retr.GetNextQueryResultBuffer2 (loop) → RetrievalService.GetNextQueryResultBuffer (loop)
|
||||
/// Retr.EndQuery2 → RetrievalService.EndQuery
|
||||
///
|
||||
/// NOTE: not yet live-verified against a 2023 R2 server. The auth handshake uses
|
||||
/// HistoryService.ExchangeKey because the gRPC HistoryService dropped ValidateClientCredential
|
||||
/// (it now lives only on StorageService) and gained ExchangeKey with the identical
|
||||
/// handle+token→token shape. If a live server rejects this, the handshake op is the first thing
|
||||
/// to revisit — everything else is the proven 2020 byte protocol.
|
||||
/// </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.");
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
HistorianDataQueryRequest request = HistorianWcfReadOrchestrator.BuildDataQueryRequest(tag, startUtc, endUtc, maxValues);
|
||||
return RunQuery(connection, clientHandle, request, maxValues, cancellationToken);
|
||||
}
|
||||
|
||||
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 RunAggregateQuery(connection, clientHandle, tag, startUtc, endUtc, mode, interval, cancellationToken);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
List<HistorianSample> results = new(timestampsUtc.Count);
|
||||
foreach (DateTime ts in timestampsUtc)
|
||||
{
|
||||
cancellationToken.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),
|
||||
cancellationToken);
|
||||
|
||||
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 uint OpenAuthenticatedConnection(HistorianGrpcConnection connection, CancellationToken cancellationToken)
|
||||
{
|
||||
Guid contextKey = Guid.NewGuid();
|
||||
var historyClient = new GrpcHistory.HistoryService.HistoryServiceClient(connection.Channel);
|
||||
|
||||
historyClient.GetInterfaceVersion(new GrpcHistory.GetInterfaceVersionRequest(), connection.Metadata, Deadline(), cancellationToken);
|
||||
|
||||
HistorianNativeHandshake.RunTokenRounds(
|
||||
(handle, wrapped, _) =>
|
||||
{
|
||||
GrpcHistory.ExchangeKeyResponse response = historyClient.ExchangeKey(
|
||||
new GrpcHistory.ExchangeKeyRequest { StrHandle = handle, BtInput = ByteString.CopyFrom(wrapped) },
|
||||
connection.Metadata,
|
||||
Deadline(),
|
||||
cancellationToken);
|
||||
byte[] serverOutput = response.BtOutput?.ToByteArray() ?? [];
|
||||
byte[] error = response.Status?.BtError?.ToByteArray() ?? [];
|
||||
bool success = response.Status?.BSuccess ?? false;
|
||||
return new HistorianNativeHandshake.TokenExchangeResult(success, serverOutput, error);
|
||||
},
|
||||
contextKey,
|
||||
_options,
|
||||
cancellationToken);
|
||||
|
||||
byte[] open2Request = HistorianNativeHandshake.BuildOpenConnection3Request(
|
||||
_options.Host, contextKey, HistorianWcfAuthChainHelper.NativeIntegratedReadOnlyConnectionMode);
|
||||
|
||||
GrpcHistory.OpenConnectionResponse open2 = historyClient.OpenConnection(
|
||||
new GrpcHistory.OpenConnectionRequest { BtConnectionRequest = ByteString.CopyFrom(open2Request) },
|
||||
connection.Metadata,
|
||||
Deadline(),
|
||||
cancellationToken);
|
||||
|
||||
byte[] open2Response = open2.BtConnectionResponse?.ToByteArray() ?? [];
|
||||
if (!(open2.Status?.BSuccess ?? false))
|
||||
{
|
||||
byte[] err = open2.Status?.BtError?.ToByteArray() ?? [];
|
||||
throw new InvalidOperationException($"gRPC OpenConnection failed (errorLen={err.Length}, responseLen={open2Response.Length}).");
|
||||
}
|
||||
|
||||
(uint clientHandle, _) = HistorianNativeHandshake.ParseOpenConnectionResponse(open2Response);
|
||||
return clientHandle;
|
||||
}
|
||||
|
||||
private List<HistorianSample> RunQuery(
|
||||
HistorianGrpcConnection connection,
|
||||
uint clientHandle,
|
||||
HistorianDataQueryRequest request,
|
||||
int maxValues,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var retrievalClient = new GrpcRetrieval.RetrievalService.RetrievalServiceClient(connection.Channel);
|
||||
retrievalClient.GetRetrievalInterfaceVersion(new GrpcRetrieval.GetRetrievalInterfaceVersionRequest(), null, Deadline(), cancellationToken);
|
||||
|
||||
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);
|
||||
retrievalClient.GetRetrievalInterfaceVersion(new GrpcRetrieval.GetRetrievalInterfaceVersionRequest(), null, Deadline(), cancellationToken);
|
||||
|
||||
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);
|
||||
}
|
||||
Reference in New Issue
Block a user