dohertj2/histsdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(grpc-events): v6 StartEventQuery request + capture-event harness scenario

Browse Source 
Captured the stock 2023 R2 client doing a gRPC event read (50 rows flowed) to
resolve the open "gRPC event ROW retrieval returns zero rows" item. Two captured
differences from our SDK's path; this lands the first (necessary) one plus the
capture tooling.

- HistorianEventQueryProtocol.CreateStartEventQueryAttempts: add a `version`
  parameter (default 5 = the 2020 WCF format, unchanged). The gRPC event
  orchestrator now opts into version 6 — the leading `06` plus a 5-byte trailing
  zero pad — which is the envelope the stock 2023 R2 client sends. The two
  buffers are otherwise byte-identical (filter block, UTC string, metadata
  namespace). Golden test Version6EmptyFilterMatchesCapturedGrpcEnvelope pins it.
- Grpc2023CaptureHarness: new `capture-event` scenario drives HistorianAccess
  over an Event-type gRPC connection (CreateEventQuery -> EventQueryArgs ->
  StartQuery -> MoveNext) so the wide-net instrument-grpc-nonstream rewrite dumps
  StartEventQuery.requestBuffer + the row result. Hostname defaults sanitized to
  HISTORIAN_GRPC_HOST / "localhost" (removed hardcoded server name).

NECESSARY BUT NOT SUFFICIENT: live validation shows v6 alone does not make rows
flow — the read also requires an Event-type connection, which our SDK's v6 Open2
format cannot express (see the companion docs commit). The gated
ReadEventsAsync_OverGrpc_* test correctly still pins the no-row throw. 322/322
offline tests pass; WCF event path unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B6mcaT2PjRFKcogzp9UkfC
This commit is contained in:
Joseph Doherty
2026-06-22 10:41:15 -04:00
parent d9051ba890
commit dbb5c99c53
4 changed files with 238 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/AVEVA.Historian.Grpc2023CaptureHarness/Program.cs
+180 -6
View File
@@ -90,8 +90,10 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
return CaptureWrite(managedDll, args);
case "delete-tag":
return DeleteTag(managedDll, args);
case "capture-event":
return CaptureEvent(managedDll, args);
default:
Console.Error.WriteLine($"Unknown scenario '{scenario}'. Supported: load-check, connect, capture-write, delete-tag.");
Console.Error.WriteLine($"Unknown scenario '{scenario}'. Supported: load-check, connect, capture-write, delete-tag, capture-event.");
return 1;
}
}
@@ -114,7 +116,7 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
Type tagStatusType = Req(asm, "ArchestrA.HistorianTagStatus");
Type tagStatusListType = Req(asm, "ArchestrA.HistorianTagStatusList");
string server = GetOption(args, "--server") ?? "WONDER-SQL-VD03";
string server = GetOption(args, "--server") ?? Environment.GetEnvironmentVariable("HISTORIAN_GRPC_HOST") ?? "localhost";
int port = int.TryParse(GetOption(args, "--port"), out int p) ? p : 32565;
string certName = GetOption(args, "--cert") ?? server;
string tagName = GetOption(args, "--tag") ?? "SdkM3CaptureSandbox";
@@ -201,7 +203,7 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
/// SendValues (the actual wire push) only runs with --commit. Run with --grpc-rewrite pointing
/// at the instrumented copy and AVEVA_HISTORIAN_RE_CAPTURE set to the output file.
/// Usage: capture-write --tag SdkM3CaptureSandbox [--create] [--commit]
/// [--server WONDER-SQL-VD03] [--port 32565] [--cert WONDER-SQL-VD03] [--value 123.0]
/// [--server <host>] [--port 32565] [--cert <host>] [--value 123.0]
/// </summary>
private static int CaptureWrite(string managedDll, string[] args)
{
@@ -221,7 +223,7 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
Type listType = Req(asm, "ArchestrA.HistorianDataValueList");
Type categoryEnum = Req(asm, "ArchestrA.HistorianDataCategory");
string server = GetOption(args, "--server") ?? "WONDER-SQL-VD03";
string server = GetOption(args, "--server") ?? Environment.GetEnvironmentVariable("HISTORIAN_GRPC_HOST") ?? "localhost";
int port = int.TryParse(GetOption(args, "--port"), out int p) ? p : 32565;
string certName = GetOption(args, "--cert") ?? server;
string tagName = GetOption(args, "--tag") ?? "SdkM3CaptureSandbox";
@@ -429,12 +431,184 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
result = m.Invoke(target, a);
}
/// <summary>
/// Drives the native 2023 R2 client through a read-only gRPC EVENT query so the IL-rewritten
/// GrpcRetrievalClient dumps the uncaptured event buffers: StartEventQuery.requestBuffer (the
/// empty-filter request shape our SDK's CreateNativeEmptyFilterAttempt is being compared against)
/// and GetNextEventQueryResultBuffer.result (the row buffer — proves rows flow when driven right).
///
/// CRITICAL: the connection is opened with ConnectionType=Event (NOT Process). CreateEventQuery()
/// returns null unless IsEventConnectionRequested() — the native event read runs on ConnectionIndex 1,
/// a separate connection from the process/data path. This is the prime suspect for why the SDK's
/// gRPC empty-filter query returns zero rows despite the server holding events.
///
/// Sequence: OpenConnection(Event, read-only, gRPC) -> CreateEventQuery() ->
/// EventQueryArgs{StartDateTime,EndDateTime,EventCount} -> EventQuery.StartQuery(args) ->
/// loop EventQuery.MoveNext()/QueryResult -> EventQuery.EndQuery() -> CloseConnection.
/// Run with --grpc-rewrite pointing at the instrumented Archestra.Historian.GrpcClient.dll and
/// AVEVA_HISTORIAN_RE_CAPTURE set to the output NDJSON. Read-only — non-destructive.
/// Usage: capture-event [--server <host>] [--port 32565] [--cert <host>]
/// [--lookback-hours 720] [--max-events 50] [--integrated]
/// </summary>
private static int CaptureEvent(string managedDll, string[] args)
{
Assembly asm = Assembly.LoadFrom(managedDll);
Type accessType = Req(asm, "ArchestrA.HistorianAccess");
Type connArgsType = Req(asm, "ArchestrA.HistorianConnectionArgs");
Type connModeType = Req(asm, "ArchestrA.HistorianConnectionMode");
Type connTypeType = Req(asm, "ArchestrA.HistorianConnectionType");
Type errorType = Req(asm, "ArchestrA.HistorianAccessError");
Type statusType = Req(asm, "ArchestrA.HistorianConnectionStatus");
Type certInfoType = Req(asm, "ArchestrA.CertificateInfo");
Type secModeType = Req(asm, "ArchestrA.HistorianSecurityMode");
Type eventQueryType = Req(asm, "ArchestrA.EventQuery");
Type eventArgsType = Req(asm, "ArchestrA.EventQueryArgs");
string server = GetOption(args, "--server") ?? Environment.GetEnvironmentVariable("HISTORIAN_GRPC_HOST") ?? "localhost";
int port = int.TryParse(GetOption(args, "--port"), out int p) ? p : 32565;
string certName = GetOption(args, "--cert") ?? server;
int lookbackHours = int.TryParse(GetOption(args, "--lookback-hours"), out int lh) ? lh : 720;
int maxEvents = int.TryParse(GetOption(args, "--max-events"), out int me) ? me : 50;
bool integrated = args.Contains("--integrated");
string? user = Environment.GetEnvironmentVariable("HISTORIAN_USER");
string? password = Environment.GetEnvironmentVariable("HISTORIAN_PASSWORD");
if (!integrated && string.IsNullOrEmpty(user))
{
Console.Error.WriteLine("Set HISTORIAN_USER/HISTORIAN_PASSWORD or pass --integrated.");
return 1;
}
object connArgs = Activator.CreateInstance(connArgsType)!;
SetProp(connArgs, "ServerName", server);
SetProp(connArgs, "TcpPort", checked((ushort)port));
SetProp(connArgs, "ConnectionMode", Enum.Parse(connModeType, "Historian")); // 2 = gRPC
SetProp(connArgs, "ConnectionType", Enum.Parse(connTypeType, "Event")); // EVENT connection
SetProp(connArgs, "ReadOnly", true);
SetProp(connArgs, "IntegratedSecurity", integrated);
SetProp(connArgs, "AllowUnTrustedConnection", true);
if (!integrated)
{
SetProp(connArgs, "UserName", user!);
SetProp(connArgs, "Password", password ?? string.Empty);
}
object certInfo = Activator.CreateInstance(certInfoType)!;
TrySetProp(certInfo, "CertificateName", certName);
TrySetProp(certInfo, "SecurityMode", Enum.Parse(secModeType, "TransportCertificate"));
TrySetProp(connArgs, "SecurityInfo", certInfo);
object access = Activator.CreateInstance(accessType)!;
object error = Activator.CreateInstance(errorType)!;
object?[] openArgs = { connArgs, error };
Console.WriteLine($"OpenConnection: server={server} port={port} mode=Historian type=Event cert={certName} integrated={integrated} readonly=true");
bool opened;
try
{
opened = (bool)accessType.GetMethod("OpenConnection", new[] { connArgsType, errorType.MakeByRefType() })!
.Invoke(access, openArgs)!;
}
catch (TargetInvocationException tie)
{
Console.Error.WriteLine($"OpenConnection threw: {tie.InnerException?.GetType().Name}: {tie.InnerException?.Message}");
return 2;
}
Console.WriteLine($"OpenConnection returned: {opened} err={DescribeError(openArgs[1])}");
if (!opened) { return 2; }
try
{
// Let the event connection (ConnectionIndex 1) come up.
MethodInfo getStatus = accessType.GetMethod("GetConnectionStatus", new[] { statusType.MakeByRefType() })
?? accessType.GetMethods().First(m => m.Name == "GetConnectionStatus" && m.GetParameters().Length == 1);
for (int i = 0; i < 10; i++)
{
object?[] sArgs = { null };
getStatus.Invoke(access, sArgs);
if (ReadBoolProp(sArgs[0], "ConnectedToServer") || !ReadBoolProp(sArgs[0], "Pending")) break;
System.Threading.Thread.Sleep(500);
}
// CreateEventQuery() is non-null only when the connection is an event connection.
MethodInfo createEventQuery = accessType.GetMethod("CreateEventQuery", Type.EmptyTypes)
?? accessType.GetMethods().First(m => m.Name == "CreateEventQuery" && m.GetParameters().Length == 0);
object? eventQuery = createEventQuery.Invoke(access, null);
Console.WriteLine($"CreateEventQuery: {(eventQuery == null ? "NULL (event connection not established!)" : "ok")}");
if (eventQuery == null) { return 3; }
// Build EventQueryArgs over the populated window. Times in UTC.
object eventArgs = Activator.CreateInstance(eventArgsType)!;
DateTime endUtc = DateTime.UtcNow;
DateTime startUtc = endUtc.AddHours(-lookbackHours);
TrySetProp(eventArgs, "StartDateTime", DateTime.SpecifyKind(startUtc, DateTimeKind.Utc));
TrySetProp(eventArgs, "EndDateTime", DateTime.SpecifyKind(endUtc, DateTimeKind.Utc));
TrySetProp(eventArgs, "EventCount", checked((uint)maxEvents));
Console.WriteLine($"EventQueryArgs: start={startUtc:o} end={endUtc:o} eventCount={maxEvents}");
// StartQuery -> triggers GrpcRetrievalClient.StartEventQuery (requestBuffer CAPTURED).
MethodInfo startQuery = eventQueryType.GetMethods()
.First(m => m.Name == "StartQuery" && m.GetParameters().Length == 2);
object?[] startArgs = { eventArgs, Activator.CreateInstance(errorType) };
bool started = (bool)startQuery.Invoke(eventQuery, startArgs)!;
Console.WriteLine($"StartQuery: {started} err={DescribeError(startArgs[1])}");
// Poll rows -> triggers GetNextEventQueryResultBuffer (result buffer CAPTURED).
MethodInfo moveNext = eventQueryType.GetMethods()
.First(m => m.Name == "MoveNext" && m.GetParameters().Length == 1);
PropertyInfo? queryResult = eventQueryType.GetProperty("QueryResult");
int rows = 0;
while (rows < maxEvents)
{
object?[] mnArgs = { Activator.CreateInstance(errorType) };
bool more;
try { more = (bool)moveNext.Invoke(eventQuery, mnArgs)!; }
catch (TargetInvocationException tie)
{
Console.WriteLine($"MoveNext threw: {tie.InnerException?.GetType().Name}: {tie.InnerException?.Message}");
break;
}
if (!more)
{
Console.WriteLine($"MoveNext: end after {rows} row(s) err={DescribeError(mnArgs[0])}");
break;
}
rows++;
if (rows <= 3 && queryResult != null)
{
// Print only the event TYPE + time (non-identity) to confirm rows flow.
object? res = queryResult.GetValue(eventQuery);
string typ = res?.GetType().GetProperty("Type")?.GetValue(res)?.ToString() ?? "?";
object? t = res?.GetType().GetProperty("EventTime")?.GetValue(res);
Console.WriteLine($" row {rows}: Type={typ} EventTime={t}");
}
}
Console.WriteLine($"Rows iterated: {rows}");
MethodInfo? endQuery = eventQueryType.GetMethods()
.FirstOrDefault(m => m.Name == "EndQuery" && m.GetParameters().Length == 1);
if (endQuery != null)
{
object?[] eqArgs = { Activator.CreateInstance(errorType) };
endQuery.Invoke(eventQuery, eqArgs);
}
Console.WriteLine(rows > 0 ? "CAPTURE-EVENT: PASS (rows flowed)" : "CAPTURE-EVENT: request captured (zero rows)");
return 0;
}
finally
{
try
{
MethodInfo? close = accessType.GetMethod("CloseConnection", new[] { errorType.MakeByRefType() });
if (close != null) close.Invoke(access, new object?[] { Activator.CreateInstance(errorType) });
}
catch { /* best-effort */ }
}
}
/// <summary>
/// Read-only gRPC connect probe: opens a 2023 R2 Historian (mode=Historian) connection via the
/// native client and reports the resulting connection status. Proves the mixed-mode client can
/// reach the live server over gRPC from this box — the foundation for the write-capture step.
/// Reads creds from HISTORIAN_USER / HISTORIAN_PASSWORD (explicit) or uses IntegratedSecurity.
/// Usage: connect --server WONDER-SQL-VD03 [--port 32565] [--cert WONDER-SQL-VD03] [--integrated]
/// Usage: connect --server <host> [--port 32565] [--cert <host>] [--integrated]
/// </summary>
private static int Connect(string managedDll, string[] args)
{
@@ -448,7 +622,7 @@ namespace AVEVA.Historian.Grpc2023CaptureHarness
Type certInfoType = Req(asm, "ArchestrA.CertificateInfo");
Type secModeType = Req(asm, "ArchestrA.HistorianSecurityMode");
string server = GetOption(args, "--server") ?? "WONDER-SQL-VD03";
string server = GetOption(args, "--server") ?? Environment.GetEnvironmentVariable("HISTORIAN_GRPC_HOST") ?? "localhost";
int port = int.TryParse(GetOption(args, "--port"), out int p) ? p : 32565;
string certName = GetOption(args, "--cert") ?? server;
bool integrated = args.Contains("--integrated");