feat(mesh-phase5): central telemetry dialer client + proto->domain converters

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-23 15:55:06 -04:00
parent e742fee452
commit c78034f0b9
5 changed files with 690 additions and 0 deletions
@@ -0,0 +1,132 @@
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
+using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
+
+/// <summary>
+///     Central-side projection of a wire <see cref="TelemetryEvent"/> envelope back onto its domain
+///     record (per-cluster mesh Phase 5). The exact reverse of <c>TelemetryProtoMapNode</c>: it
+///     transcribes every proto field onto the matching domain field and selects the domain record from
+///     the envelope's <c>oneof</c> arm.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Timestamp presence is honoured.</b> A proto <c>google.protobuf.Timestamp</c> field is a
+///         message: when the node left it unset (a null nullable-<c>DateTime</c> on the domain side) the
+///         generated property is <see langword="null"/>, so the nullable domain fields map via
+///         <c>?.ToDateTime()</c> and an absent Timestamp becomes <see langword="null"/>. The
+///         non-nullable domain fields (published / sampled / transition timestamps) are always set by
+///         the node mapper, so they map via a direct <c>.ToDateTime()</c> (which yields a
+///         <see cref="DateTimeKind.Utc"/> value).
+///     </para>
+///     <para>
+///         <b>optional-string presence is honoured.</b> A proto3 <c>optional string</c> that the node
+///         left unset reads back as <see langword="null"/> (via the generated <c>Has…</c> presence),
+///         distinguishing null from the empty string; likewise the <c>optional bool</c>
+///         <c>historize_to_aveva</c> round-trips its three states (null / true / false).
+///     </para>
+/// </remarks>
+public static class TelemetryProtoMapCentral
+{
+    /// <summary>
+    ///     Projects a wire <see cref="TelemetryEvent"/> onto its domain record, selecting the type from
+    ///     the populated <c>oneof</c> arm. This switch is the coverage guard: every
+    ///     <see cref="TelemetryProtoContract.HandledCases"/> value has an arm here, and an unset (or a
+    ///     future unmapped) case throws <see cref="NotSupportedException"/> so adding a fifth oneof case
+    ///     without a converter fails the coverage test.
+    /// </summary>
+    /// <param name="evt">The wire envelope to project.</param>
+    /// <returns>The domain record for the populated arm.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="evt"/> is null.</exception>
+    /// <exception cref="NotSupportedException">The envelope carries no handled <c>oneof</c> arm.</exception>
+    public static object MapEvent(TelemetryEvent evt)
+    {
+        ArgumentNullException.ThrowIfNull(evt);
+
+        return evt.EventCase switch
+        {
+            TelemetryEvent.EventOneofCase.AlarmTransition => ToAlarm(evt.AlarmTransition),
+            TelemetryEvent.EventOneofCase.ScriptLog => ToScript(evt.ScriptLog),
+            TelemetryEvent.EventOneofCase.DriverHealth => ToHealth(evt.DriverHealth),
+            TelemetryEvent.EventOneofCase.DriverResilience => ToResilience(evt.DriverResilience),
+            _ => throw new NotSupportedException(
+                $"TelemetryEvent carries no handled oneof arm (EventCase = {evt.EventCase})."),
+        };
+    }
+
+    /// <summary>Projects an <see cref="AlarmTransition"/> onto an <see cref="AlarmTransitionEvent"/>.</summary>
+    /// <param name="msg">The proto sub-message.</param>
+    /// <returns>The domain record.</returns>
+    public static AlarmTransitionEvent ToAlarm(AlarmTransition msg)
+    {
+        ArgumentNullException.ThrowIfNull(msg);
+
+        return new AlarmTransitionEvent(
+            AlarmId: msg.AlarmId,
+            EquipmentPath: msg.EquipmentPath,
+            AlarmName: msg.AlarmName,
+            TransitionKind: msg.TransitionKind,
+            Severity: msg.Severity,
+            Message: msg.Message,
+            User: msg.User,
+            TimestampUtc: msg.TimestampUtc.ToDateTime(),
+            AlarmTypeName: msg.AlarmTypeName,
+            Comment: msg.HasComment ? msg.Comment : null,
+            HistorizeToAveva: msg.HasHistorizeToAveva ? msg.HistorizeToAveva : null,
+            ReferencingEquipmentPaths: msg.ReferencingEquipmentPaths.ToList());
+    }
+
+    /// <summary>Projects a <see cref="ScriptLog"/> onto a <see cref="ScriptLogEntry"/>.</summary>
+    /// <param name="msg">The proto sub-message.</param>
+    /// <returns>The domain record.</returns>
+    public static ScriptLogEntry ToScript(ScriptLog msg)
+    {
+        ArgumentNullException.ThrowIfNull(msg);
+
+        return new ScriptLogEntry(
+            ScriptId: msg.ScriptId,
+            Level: msg.Level,
+            Message: msg.Message,
+            TimestampUtc: msg.TimestampUtc.ToDateTime(),
+            VirtualTagId: msg.HasVirtualTagId ? msg.VirtualTagId : null,
+            AlarmId: msg.HasAlarmId ? msg.AlarmId : null,
+            EquipmentId: msg.HasEquipmentId ? msg.EquipmentId : null);
+    }
+
+    /// <summary>Projects a <see cref="DriverHealth"/> onto a <see cref="DriverHealthChanged"/>.</summary>
+    /// <param name="msg">The proto sub-message.</param>
+    /// <returns>The domain record.</returns>
+    public static DriverHealthChanged ToHealth(DriverHealth msg)
+    {
+        ArgumentNullException.ThrowIfNull(msg);
+
+        return new DriverHealthChanged(
+            ClusterId: msg.ClusterId,
+            DriverInstanceId: msg.DriverInstanceId,
+            State: msg.State,
+            LastSuccessfulReadUtc: msg.LastSuccessfulReadUtc?.ToDateTime(),
+            LastError: msg.HasLastError ? msg.LastError : null,
+            ErrorCount5Min: msg.ErrorCount5Min,
+            PublishedUtc: msg.PublishedUtc.ToDateTime());
+    }
+
+    /// <summary>Projects a <see cref="DriverResilienceStatus"/> onto a <see cref="DriverResilienceStatusChanged"/>.</summary>
+    /// <param name="msg">The proto sub-message.</param>
+    /// <returns>The domain record.</returns>
+    public static DriverResilienceStatusChanged ToResilience(DriverResilienceStatus msg)
+    {
+        ArgumentNullException.ThrowIfNull(msg);
+
+        return new DriverResilienceStatusChanged(
+            DriverInstanceId: msg.DriverInstanceId,
+            HostName: msg.HostName,
+            BreakerOpen: msg.BreakerOpen,
+            ConsecutiveFailures: msg.ConsecutiveFailures,
+            CurrentInFlight: msg.CurrentInFlight,
+            LastBreakerOpenUtc: msg.LastBreakerOpenUtc?.ToDateTime(),
+            LastSampledUtc: msg.LastSampledUtc.ToDateTime(),
+            PublishedUtc: msg.PublishedUtc.ToDateTime());
+    }
+}
@@ -0,0 +1,123 @@
+using System.Net;
+using Grpc.Core;
+using Grpc.Net.Client;
+using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
+
+/// <summary>
+///     Central-side per-node dialer for the Phase 5 telemetry stream. Owns one h2c
+///     <see cref="GrpcChannel"/> to a single driver node's <c>TelemetryStreamService</c>, opens the
+///     server-streaming <c>Subscribe</c> call, and pumps every <see cref="TelemetryEvent"/> to a
+///     caller-supplied sink. The raw proto envelope is delivered as-is — the caller maps it via
+///     <see cref="TelemetryProtoMapCentral"/>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Deliberately dumb.</b> Connect, pump, surface errors — no reconnect, no backoff, no
+///         buffering. A normal shutdown (the caller's <see cref="CancellationToken"/> firing, or the
+///         server ending the stream with <see cref="StatusCode.Cancelled"/>) completes
+///         <see cref="RunAsync"/> without calling <c>onError</c>; any other failure is routed to
+///         <c>onError</c> exactly once as the reconnect trigger the supervisor (a later task) owns.
+///     </para>
+///     <para>
+///         <b>h2c + Bearer.</b> The endpoint is a prior-knowledge <c>http://host:port</c> address (no
+///         TLS); the shared node bearer key rides the <c>authorization</c> header on every call. The
+///         channel sets HTTP/2 keepalive pings so a long-idle telemetry stream is not silently dropped
+///         by an intermediary.
+///     </para>
+/// </remarks>
+public sealed class TelemetryStreamClient : IDisposable
+{
+    private readonly string _apiKey;
+    private readonly GrpcChannel? _ownedChannel;
+    private readonly TelemetryStreamService.TelemetryStreamServiceClient _client;
+
+    /// <summary>
+    ///     Initializes a new instance of the <see cref="TelemetryStreamClient"/> class dialing a live
+    ///     driver node over h2c.
+    /// </summary>
+    /// <param name="endpoint">Prior-knowledge <c>http://host:port</c> address of the node's telemetry server.</param>
+    /// <param name="apiKey">Shared node bearer key sent as <c>authorization: Bearer &lt;key&gt;</c>.</param>
+    public TelemetryStreamClient(string endpoint, string apiKey)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(endpoint);
+        ArgumentNullException.ThrowIfNull(apiKey);
+
+        _apiKey = apiKey;
+        _ownedChannel = GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions
+        {
+            HttpHandler = new SocketsHttpHandler
+            {
+                KeepAlivePingDelay = TimeSpan.FromSeconds(15),
+                KeepAlivePingTimeout = TimeSpan.FromSeconds(10),
+                KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always,
+                EnableMultipleHttp2Connections = true,
+            },
+        });
+        _client = new TelemetryStreamService.TelemetryStreamServiceClient(_ownedChannel);
+    }
+
+    /// <summary>
+    ///     Initializes a new instance of the <see cref="TelemetryStreamClient"/> class over a
+    ///     caller-supplied client — the test seam. No channel is owned or disposed.
+    /// </summary>
+    /// <param name="client">The generated client to drive (typically a fake in tests).</param>
+    /// <param name="apiKey">Shared node bearer key sent as <c>authorization: Bearer &lt;key&gt;</c>.</param>
+    public TelemetryStreamClient(TelemetryStreamService.TelemetryStreamServiceClient client, string apiKey)
+    {
+        ArgumentNullException.ThrowIfNull(client);
+        ArgumentNullException.ThrowIfNull(apiKey);
+
+        _client = client;
+        _apiKey = apiKey;
+        _ownedChannel = null;
+    }
+
+    /// <summary>
+    ///     Opens the telemetry stream and pumps every envelope to <paramref name="onEvent"/> until the
+    ///     server ends the stream or <paramref name="ct"/> fires. A normal shutdown returns without
+    ///     invoking <paramref name="onError"/>; any other failure invokes <paramref name="onError"/>
+    ///     exactly once and returns.
+    /// </summary>
+    /// <param name="correlationId">Stream correlation id echoed on every envelope.</param>
+    /// <param name="onEvent">Sink for each received raw <see cref="TelemetryEvent"/>.</param>
+    /// <param name="onError">Invoked once on an abnormal stream failure (the reconnect trigger).</param>
+    /// <param name="ct">Cancels the stream (normal shutdown).</param>
+    public async Task RunAsync(
+        string correlationId,
+        Action<TelemetryEvent> onEvent,
+        Action<Exception> onError,
+        CancellationToken ct)
+    {
+        ArgumentNullException.ThrowIfNull(onEvent);
+        ArgumentNullException.ThrowIfNull(onError);
+
+        var headers = new Metadata { { "authorization", $"Bearer {_apiKey}" } };
+
+        try
+        {
+            using var call = _client.Subscribe(
+                new TelemetryStreamRequest { CorrelationId = correlationId }, headers, cancellationToken: ct);
+
+            await foreach (var evt in call.ResponseStream.ReadAllAsync(ct))
+                onEvent(evt);
+        }
+        catch (OperationCanceledException)
+        {
+            // Normal shutdown: the caller's token fired.
+        }
+        catch (RpcException rex) when (rex.StatusCode == StatusCode.Cancelled)
+        {
+            // Normal shutdown: the server (or a token-driven cancel) ended the stream.
+        }
+        catch (Exception ex)
+        {
+            // The reconnect trigger — a later supervisor task decides retry/backoff.
+            onError(ex);
+        }
+    }
+
+    /// <inheritdoc />
+    public void Dispose() => _ownedChannel?.Dispose();
+}
@@ -15,6 +15,9 @@
        <PackageReference Include="Akka.Cluster.Hosting"/>
        <PackageReference Include="Akka.Cluster.Tools"/>
        <PackageReference Include="Microsoft.EntityFrameworkCore"/>
+        <!-- Phase 5: central dials each driver node's telemetry gRPC server (GrpcChannel h2c);
+             Google.Protobuf + Grpc.Core.Api flow transitively from Commons (the generated client). -->
+        <PackageReference Include="Grpc.Net.Client"/>
        <PackageReference Include="ZB.MOM.WW.Audit"/>
    </ItemGroup>