feat(mesh-phase5): central telemetry dial supervisor (discover + reconnect + feed sinks)
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,468 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Event;
|
||||
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
|
||||
using ZB.MOM.WW.OtOpcUa.Cluster;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
|
||||
|
||||
/// <summary>
|
||||
/// A single dialable driver node: its logical id and the prior-knowledge
|
||||
/// <c>http://host:port</c> gRPC telemetry endpoint central connects to.
|
||||
/// </summary>
|
||||
/// <param name="NodeId">The node's stable logical id (the <c>ClusterNode.NodeId</c>).</param>
|
||||
/// <param name="Endpoint">The <c>http://{Host}:{GrpcPort}</c> h2c telemetry endpoint.</param>
|
||||
public sealed record TelemetryDialTarget(string NodeId, string Endpoint);
|
||||
|
||||
/// <summary>
|
||||
/// The dial-loop seam: opens one telemetry stream to <paramref name="target"/> and pumps mapped
|
||||
/// domain records to <paramref name="onMapped"/> until the stream ends or <paramref name="ct"/>
|
||||
/// fires, routing a transport failure to <paramref name="onError"/>. The production
|
||||
/// implementation wraps <c>TelemetryStreamClient</c> + <c>TelemetryProtoMapCentral</c>; tests
|
||||
/// substitute a fake that captures the callbacks and drives events/failures by hand.
|
||||
/// </summary>
|
||||
/// <param name="target">The node to dial.</param>
|
||||
/// <param name="correlationId">Per-generation stream correlation id.</param>
|
||||
/// <param name="onMapped">Sink for each mapped domain record (already projected from the wire).</param>
|
||||
/// <param name="onError">Invoked on a transport failure (the reconnect trigger).</param>
|
||||
/// <param name="ct">Cancels the stream (a normal, supervisor-initiated shutdown).</param>
|
||||
/// <returns>A task that completes when the stream ends or is cancelled.</returns>
|
||||
public delegate Task TelemetryDialLoop(
|
||||
TelemetryDialTarget target,
|
||||
string correlationId,
|
||||
Action<object> onMapped,
|
||||
Action<Exception> onError,
|
||||
CancellationToken ct);
|
||||
|
||||
/// <summary>
|
||||
/// Central-side supervisor of the Phase 5 telemetry dials. Maintains one reconnecting dialer per
|
||||
/// enabled, non-maintenance driver node and feeds every received record to the same four AdminUI
|
||||
/// in-process sinks the DPS bridges feed today — so flipping central from
|
||||
/// <c>TelemetryDial:Mode = Dps</c> to <c>Grpc</c> is transparent to the Blazor panels.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Discover → dial → reconnect, forever.</b> On a periodic refresh the supervisor resolves
|
||||
/// the current dialable node set (via the injected <see cref="Func{TResult}"/> node source),
|
||||
/// starts a dialer for each new node id, and stops+removes a dialer whose node id has left the
|
||||
/// set. A dialer whose stream drops reconnects indefinitely (first retry immediate, then a
|
||||
/// fixed backoff) — this is an observability plane, not a data plane, so it never gives up.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>All state mutation is on the actor thread.</b> The background dialer tasks only ever
|
||||
/// <c>Self.Tell</c> — they never touch the dialer table, the connection pill, or a generation
|
||||
/// counter directly. Each dialer carries a monotonic <em>generation</em>; a message tagged with
|
||||
/// a stale generation (a late event or a late failure from a superseded stream) is dropped, so a
|
||||
/// reconnect can neither double-fire nor route a stale record.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
|
||||
{
|
||||
private const string RefreshTimerKey = "telemetry-node-refresh";
|
||||
private static readonly TimeSpan ReconnectBackoff = TimeSpan.FromSeconds(5);
|
||||
|
||||
private readonly Func<Task<IReadOnlyList<TelemetryDialTarget>>> _nodeSource;
|
||||
private readonly TelemetryDialLoop _dialLoop;
|
||||
private readonly IInProcessBroadcaster<AlarmTransitionEvent> _alarmBroadcaster;
|
||||
private readonly IInProcessBroadcaster<ScriptLogEntry> _scriptBroadcaster;
|
||||
private readonly IDriverStatusSnapshotStore _healthStore;
|
||||
private readonly IDriverResilienceStatusStore _resilienceStore;
|
||||
private readonly TelemetryDialOptions _options;
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
|
||||
private readonly Dictionary<string, NodeDialer> _dialers = new(StringComparer.Ordinal);
|
||||
private bool _pillConnected;
|
||||
|
||||
/// <summary>Gets the timer scheduler driving the periodic refresh and per-node reconnect backoff.</summary>
|
||||
public ITimerScheduler Timers { get; set; } = null!;
|
||||
|
||||
/// <summary>Creates the props for the telemetry dial supervisor.</summary>
|
||||
/// <param name="nodeSource">Resolves the current dialable node set (off the actor thread).</param>
|
||||
/// <param name="dialLoop">Opens + pumps one node's telemetry stream (the reconnect unit).</param>
|
||||
/// <param name="alarmBroadcaster">Sink for alarm-transition records.</param>
|
||||
/// <param name="scriptBroadcaster">Sink for script-log records.</param>
|
||||
/// <param name="healthStore">Sink for driver-health snapshots.</param>
|
||||
/// <param name="resilienceStore">Sink for driver-resilience snapshots.</param>
|
||||
/// <param name="options">The bound central dial options (refresh cadence etc.).</param>
|
||||
/// <returns>The props.</returns>
|
||||
public static Props Props(
|
||||
Func<Task<IReadOnlyList<TelemetryDialTarget>>> nodeSource,
|
||||
TelemetryDialLoop dialLoop,
|
||||
IInProcessBroadcaster<AlarmTransitionEvent> alarmBroadcaster,
|
||||
IInProcessBroadcaster<ScriptLogEntry> scriptBroadcaster,
|
||||
IDriverStatusSnapshotStore healthStore,
|
||||
IDriverResilienceStatusStore resilienceStore,
|
||||
TelemetryDialOptions options) =>
|
||||
Akka.Actor.Props.Create(() => new TelemetryDialSupervisor(
|
||||
nodeSource, dialLoop, alarmBroadcaster, scriptBroadcaster, healthStore, resilienceStore, options));
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="TelemetryDialSupervisor"/> class.</summary>
|
||||
/// <param name="nodeSource">Resolves the current dialable node set (off the actor thread).</param>
|
||||
/// <param name="dialLoop">Opens + pumps one node's telemetry stream (the reconnect unit).</param>
|
||||
/// <param name="alarmBroadcaster">Sink for alarm-transition records.</param>
|
||||
/// <param name="scriptBroadcaster">Sink for script-log records.</param>
|
||||
/// <param name="healthStore">Sink for driver-health snapshots.</param>
|
||||
/// <param name="resilienceStore">Sink for driver-resilience snapshots.</param>
|
||||
/// <param name="options">The bound central dial options (refresh cadence etc.).</param>
|
||||
public TelemetryDialSupervisor(
|
||||
Func<Task<IReadOnlyList<TelemetryDialTarget>>> nodeSource,
|
||||
TelemetryDialLoop dialLoop,
|
||||
IInProcessBroadcaster<AlarmTransitionEvent> alarmBroadcaster,
|
||||
IInProcessBroadcaster<ScriptLogEntry> scriptBroadcaster,
|
||||
IDriverStatusSnapshotStore healthStore,
|
||||
IDriverResilienceStatusStore resilienceStore,
|
||||
TelemetryDialOptions options)
|
||||
{
|
||||
_nodeSource = nodeSource;
|
||||
_dialLoop = dialLoop;
|
||||
_alarmBroadcaster = alarmBroadcaster;
|
||||
_scriptBroadcaster = scriptBroadcaster;
|
||||
_healthStore = healthStore;
|
||||
_resilienceStore = resilienceStore;
|
||||
_options = options;
|
||||
|
||||
Receive<RefreshNodes>(_ => HandleRefreshNodes());
|
||||
Receive<NodesResolved>(HandleNodesResolved);
|
||||
Receive<TelemetryReceived>(HandleTelemetryReceived);
|
||||
Receive<StreamStopped>(HandleStreamStopped);
|
||||
Receive<Reconnect>(HandleReconnect);
|
||||
|
||||
// A faulted node-source task pipes here. Log and wait for the next periodic refresh — a stale
|
||||
// node set is far better than crashing the supervisor and dropping every live dialer.
|
||||
Receive<Status.Failure>(f => _log.Warning(
|
||||
f.Cause,
|
||||
"Telemetry node source failed; the dialable node set was NOT refreshed and may be stale. "
|
||||
+ "Existing dialers keep running; the next periodic refresh retries"));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PreStart()
|
||||
{
|
||||
Timers.StartPeriodicTimer(
|
||||
RefreshTimerKey,
|
||||
new RefreshNodes(),
|
||||
TimeSpan.Zero,
|
||||
TimeSpan.FromSeconds(Math.Max(1, _options.ContactRefreshSeconds)));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
foreach (var dialer in _dialers.Values)
|
||||
{
|
||||
dialer.Cancel();
|
||||
}
|
||||
|
||||
_dialers.Clear();
|
||||
}
|
||||
|
||||
private void HandleRefreshNodes()
|
||||
{
|
||||
Task<IReadOnlyList<TelemetryDialTarget>> task;
|
||||
try
|
||||
{
|
||||
task = _nodeSource();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(ex, "Telemetry node source threw synchronously; skipping this refresh");
|
||||
return;
|
||||
}
|
||||
|
||||
task.PipeTo(
|
||||
Self,
|
||||
success: targets => new NodesResolved(targets),
|
||||
failure: ex => new Status.Failure(ex));
|
||||
}
|
||||
|
||||
private void HandleNodesResolved(NodesResolved msg)
|
||||
{
|
||||
var incoming = new Dictionary<string, TelemetryDialTarget>(StringComparer.Ordinal);
|
||||
foreach (var target in msg.Targets)
|
||||
{
|
||||
// Last write wins on a duplicate id; a malformed set is the node source's problem, not ours.
|
||||
incoming[target.NodeId] = target;
|
||||
}
|
||||
|
||||
// Drop dialers whose node is no longer present.
|
||||
foreach (var nodeId in _dialers.Keys.ToList())
|
||||
{
|
||||
if (!incoming.ContainsKey(nodeId))
|
||||
{
|
||||
StopDialer(nodeId);
|
||||
}
|
||||
}
|
||||
|
||||
// Start a dialer for each newly-present node. Existing dialers keep running untouched.
|
||||
foreach (var (nodeId, target) in incoming)
|
||||
{
|
||||
if (!_dialers.ContainsKey(nodeId))
|
||||
{
|
||||
var dialer = new NodeDialer(target);
|
||||
_dialers[nodeId] = dialer;
|
||||
StartDialer(nodeId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleTelemetryReceived(TelemetryReceived msg)
|
||||
{
|
||||
if (!_dialers.TryGetValue(msg.NodeId, out var dialer) || msg.Generation != dialer.Generation)
|
||||
{
|
||||
// Superseded stream (a reconnect already advanced the generation) — drop, never route.
|
||||
return;
|
||||
}
|
||||
|
||||
if (!dialer.Connected)
|
||||
{
|
||||
dialer.Connected = true;
|
||||
dialer.FailureStreak = 0;
|
||||
RecomputePill();
|
||||
}
|
||||
|
||||
Route(msg.NodeId, msg.Record);
|
||||
}
|
||||
|
||||
private void HandleStreamStopped(StreamStopped msg)
|
||||
{
|
||||
if (!_dialers.TryGetValue(msg.NodeId, out var dialer) || msg.Generation != dialer.Generation)
|
||||
{
|
||||
// Late failure/end from a superseded stream — ignore, so a reconnect never double-fires.
|
||||
return;
|
||||
}
|
||||
|
||||
if (dialer.Connected)
|
||||
{
|
||||
dialer.Connected = false;
|
||||
RecomputePill();
|
||||
}
|
||||
|
||||
dialer.FailureStreak++;
|
||||
|
||||
if (msg.Error is not null && (dialer.FailureStreak == 1 || dialer.FailureStreak % 10 == 0))
|
||||
{
|
||||
_log.Warning(
|
||||
msg.Error,
|
||||
"Telemetry stream to node {NodeId} failed (failure streak {Streak}); reconnecting",
|
||||
msg.NodeId, dialer.FailureStreak);
|
||||
}
|
||||
else
|
||||
{
|
||||
_log.Debug(
|
||||
"Telemetry stream to node {NodeId} stopped (failure streak {Streak}); reconnecting",
|
||||
msg.NodeId, dialer.FailureStreak);
|
||||
}
|
||||
|
||||
if (dialer.FailureStreak == 1)
|
||||
{
|
||||
// First retry after a healthy stream is immediate.
|
||||
StartDialer(msg.NodeId);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Subsequent retries back off. A single timer per node — a queued reconnect is replaced.
|
||||
Timers.StartSingleTimer(ReconnectTimerKey(msg.NodeId), new Reconnect(msg.NodeId), ReconnectBackoff);
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleReconnect(Reconnect msg)
|
||||
{
|
||||
// The node may have been removed while the backoff timer was pending.
|
||||
if (_dialers.ContainsKey(msg.NodeId))
|
||||
{
|
||||
StartDialer(msg.NodeId);
|
||||
}
|
||||
}
|
||||
|
||||
private void StartDialer(string nodeId)
|
||||
{
|
||||
var dialer = _dialers[nodeId];
|
||||
|
||||
// Cancel any prior stream and advance the generation so its in-flight signals are ignored.
|
||||
dialer.Cancel();
|
||||
Timers.Cancel(ReconnectTimerKey(nodeId));
|
||||
|
||||
var generation = ++dialer.Generation;
|
||||
var cts = new CancellationTokenSource();
|
||||
dialer.Cts = cts;
|
||||
dialer.Connected = false;
|
||||
|
||||
var self = Self;
|
||||
var target = dialer.Target;
|
||||
var token = cts.Token;
|
||||
var correlationId = $"central-{nodeId}-{generation}";
|
||||
var dialLoop = _dialLoop;
|
||||
|
||||
// The background task ONLY ever Self.Tell()s — it never touches actor state.
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
var faulted = false;
|
||||
try
|
||||
{
|
||||
await dialLoop(
|
||||
target,
|
||||
correlationId,
|
||||
onMapped: record => self.Tell(new TelemetryReceived(nodeId, generation, record)),
|
||||
onError: error =>
|
||||
{
|
||||
faulted = true;
|
||||
self.Tell(new StreamStopped(nodeId, generation, error));
|
||||
},
|
||||
token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Normal shutdown: our cancellation token fired. No reconnect.
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// The dial loop threw instead of routing to onError — treat as a transport failure.
|
||||
self.Tell(new StreamStopped(nodeId, generation, ex));
|
||||
return;
|
||||
}
|
||||
|
||||
// The loop returned. If it did not already fault and we did not cancel it, the server ended
|
||||
// the stream — reconnect. (A faulted loop already told StreamStopped with the error.)
|
||||
if (!faulted && !token.IsCancellationRequested)
|
||||
{
|
||||
self.Tell(new StreamStopped(nodeId, generation, Error: null));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void StopDialer(string nodeId)
|
||||
{
|
||||
if (_dialers.Remove(nodeId, out var dialer))
|
||||
{
|
||||
dialer.Cancel();
|
||||
Timers.Cancel(ReconnectTimerKey(nodeId));
|
||||
RecomputePill();
|
||||
}
|
||||
}
|
||||
|
||||
private void Route(string nodeId, object record)
|
||||
{
|
||||
// Defensive: one poison record must never crash the supervisor and take every dialer with it.
|
||||
try
|
||||
{
|
||||
switch (record)
|
||||
{
|
||||
case AlarmTransitionEvent alarm:
|
||||
_alarmBroadcaster.Publish(alarm);
|
||||
break;
|
||||
case ScriptLogEntry script:
|
||||
_scriptBroadcaster.Publish(script);
|
||||
break;
|
||||
case DriverHealthChanged health:
|
||||
_healthStore.Upsert(health);
|
||||
break;
|
||||
case DriverResilienceStatusChanged resilience:
|
||||
_resilienceStore.Upsert(resilience);
|
||||
break;
|
||||
default:
|
||||
_log.Warning(
|
||||
"Telemetry record of unrouted type {RecordType} from node {NodeId}; dropped",
|
||||
record?.GetType().Name ?? "null", nodeId);
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(
|
||||
ex,
|
||||
"Failed to route a telemetry record of type {RecordType} from node {NodeId} to its sink",
|
||||
record?.GetType().Name ?? "null", nodeId);
|
||||
}
|
||||
}
|
||||
|
||||
private void RecomputePill()
|
||||
{
|
||||
var anyConnected = false;
|
||||
foreach (var dialer in _dialers.Values)
|
||||
{
|
||||
if (dialer.Connected)
|
||||
{
|
||||
anyConnected = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (anyConnected == _pillConnected)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_pillConnected = anyConnected;
|
||||
_alarmBroadcaster.SetConnected(anyConnected);
|
||||
_scriptBroadcaster.SetConnected(anyConnected);
|
||||
}
|
||||
|
||||
private static string ReconnectTimerKey(string nodeId) => $"telemetry-reconnect-{nodeId}";
|
||||
|
||||
/// <summary>Per-node dialer bookkeeping. Mutated only on the actor thread.</summary>
|
||||
private sealed class NodeDialer(TelemetryDialTarget target)
|
||||
{
|
||||
/// <summary>The node's dial target.</summary>
|
||||
public TelemetryDialTarget Target { get; } = target;
|
||||
|
||||
/// <summary>Monotonic stream generation; a message tagged with a stale value is dropped.</summary>
|
||||
public int Generation { get; set; }
|
||||
|
||||
/// <summary>Cancels the current stream's background task.</summary>
|
||||
public CancellationTokenSource? Cts { get; set; }
|
||||
|
||||
/// <summary>Whether the current stream has delivered at least one event.</summary>
|
||||
public bool Connected { get; set; }
|
||||
|
||||
/// <summary>Consecutive stream stops since the last successful event (drives immediate vs. backoff).</summary>
|
||||
public int FailureStreak { get; set; }
|
||||
|
||||
/// <summary>Cancels + disposes the current stream's token source, if any.</summary>
|
||||
public void Cancel()
|
||||
{
|
||||
var cts = Cts;
|
||||
Cts = null;
|
||||
if (cts is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
cts.Cancel();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// Already disposed — nothing to cancel.
|
||||
}
|
||||
|
||||
cts.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Timer tick: re-resolve the dialable node set.</summary>
|
||||
public sealed record RefreshNodes;
|
||||
|
||||
/// <summary>The node set resolved off the actor thread.</summary>
|
||||
/// <param name="Targets">The current dialable nodes.</param>
|
||||
public sealed record NodesResolved(IReadOnlyList<TelemetryDialTarget> Targets);
|
||||
|
||||
/// <summary>A mapped domain record received from a node's stream.</summary>
|
||||
/// <param name="NodeId">The source node id.</param>
|
||||
/// <param name="Generation">The dialer generation that produced it.</param>
|
||||
/// <param name="Record">The mapped domain record.</param>
|
||||
public sealed record TelemetryReceived(string NodeId, int Generation, object Record);
|
||||
|
||||
/// <summary>A node's stream stopped — transport failure (<paramref name="Error"/> set) or server-ended.</summary>
|
||||
/// <param name="NodeId">The node id whose stream stopped.</param>
|
||||
/// <param name="Generation">The dialer generation that stopped.</param>
|
||||
/// <param name="Error">The transport failure, or <see langword="null"/> when the server ended the stream.</param>
|
||||
public sealed record StreamStopped(string NodeId, int Generation, Exception? Error);
|
||||
|
||||
/// <summary>Backoff timer tick: start a fresh dialer for the node.</summary>
|
||||
/// <param name="NodeId">The node id to reconnect.</param>
|
||||
public sealed record Reconnect(string NodeId);
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
|
||||
|
||||
/// <summary>
|
||||
/// Production seams for <see cref="TelemetryDialSupervisor"/>: the DB-backed node source and the
|
||||
/// gRPC dial loop. Kept out of the actor so the actor stays DB-free and gRPC-free (and therefore
|
||||
/// unit-testable with fakes).
|
||||
/// </summary>
|
||||
public static class TelemetryNodeSource
|
||||
{
|
||||
/// <summary>
|
||||
/// Builds the node source the supervisor polls: enabled, non-maintenance <c>ClusterNode</c>
|
||||
/// rows carrying a non-null <c>GrpcPort</c>, projected to <c>http://{Host}:{GrpcPort}</c>
|
||||
/// h2c endpoints. A row with a null <c>GrpcPort</c> exposes no telemetry surface yet and is
|
||||
/// skipped with a Warning — mirrors <c>CentralCommunicationActor.LoadContactsFromDb</c>'s
|
||||
/// enabled + non-maintenance read.
|
||||
/// </summary>
|
||||
/// <param name="dbFactory">Factory for the config database holding <c>ClusterNode</c> rows.</param>
|
||||
/// <param name="logger">Logger for skipped-row diagnostics.</param>
|
||||
/// <returns>An async delegate resolving the current dialable node set.</returns>
|
||||
public static Func<Task<IReadOnlyList<TelemetryDialTarget>>> Create(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
||||
ILogger logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(dbFactory);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
return async () =>
|
||||
{
|
||||
await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false);
|
||||
var rows = await db.ClusterNodes
|
||||
.AsNoTracking()
|
||||
.Where(n => n.Enabled && !n.MaintenanceMode)
|
||||
.Select(n => new { n.NodeId, n.Host, n.GrpcPort })
|
||||
.ToListAsync()
|
||||
.ConfigureAwait(false);
|
||||
|
||||
var targets = new List<TelemetryDialTarget>(rows.Count);
|
||||
foreach (var row in rows)
|
||||
{
|
||||
if (row.GrpcPort is null)
|
||||
{
|
||||
logger.LogWarning(
|
||||
"ClusterNode {NodeId} has no GrpcPort; it exposes no telemetry stream and is "
|
||||
+ "skipped in this dial refresh", row.NodeId);
|
||||
continue;
|
||||
}
|
||||
|
||||
targets.Add(new TelemetryDialTarget(row.NodeId, $"http://{row.Host}:{row.GrpcPort}"));
|
||||
}
|
||||
|
||||
return (IReadOnlyList<TelemetryDialTarget>)targets;
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds the production dial loop: one <see cref="TelemetryStreamClient"/> per stream, mapping
|
||||
/// every wire envelope via <see cref="TelemetryProtoMapCentral.MapEvent"/> before handing it to
|
||||
/// the supervisor. The client isolates a mapper throw (logs + continues) and routes only
|
||||
/// transport failures to <c>onError</c>, so the supervisor sees a poison event as "dropped",
|
||||
/// never as "node down".
|
||||
/// </summary>
|
||||
/// <param name="apiKey">Shared node bearer key sent on every dial.</param>
|
||||
/// <param name="clientLogger">Logger passed to each per-stream client (dropped-event diagnostics).</param>
|
||||
/// <returns>The dial-loop delegate the supervisor invokes per generation.</returns>
|
||||
public static TelemetryDialLoop CreateDialLoop(
|
||||
string apiKey,
|
||||
ILogger<TelemetryStreamClient>? clientLogger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(apiKey);
|
||||
|
||||
return async (target, correlationId, onMapped, onError, ct) =>
|
||||
{
|
||||
using var client = new TelemetryStreamClient(target.Endpoint, apiKey, clientLogger);
|
||||
await client.RunAsync(
|
||||
correlationId,
|
||||
onEvent: evt => onMapped(TelemetryProtoMapCentral.MapEvent(evt)),
|
||||
onError: onError,
|
||||
ct).ConfigureAwait(false);
|
||||
};
|
||||
}
|
||||
}
|
||||
+433
@@ -0,0 +1,433 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
|
||||
using ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
|
||||
using ZB.MOM.WW.OtOpcUa.Cluster;
|
||||
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.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry;
|
||||
|
||||
/// <summary>
|
||||
/// Unit tests for <see cref="TelemetryDialSupervisor"/> — the central-side dial supervisor that
|
||||
/// discovers driver nodes, keeps one reconnecting dialer each, and feeds the four AdminUI
|
||||
/// in-process sinks. All seams are faked: no real DB, no real gRPC.
|
||||
/// </summary>
|
||||
public sealed class TelemetryDialSupervisorTests : TestKit
|
||||
{
|
||||
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
[Fact]
|
||||
public void Discovery_starts_one_dialer_per_target_with_distinct_correlation_ids()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var targets = new[]
|
||||
{
|
||||
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
||||
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
||||
};
|
||||
|
||||
Spawn(dial, () => targets);
|
||||
|
||||
var invocations = dial.WaitForCount(2, Timeout);
|
||||
invocations.Select(i => i.Target.NodeId).ShouldBe(new[] { "node-1", "node-2" }, ignoreOrder: true);
|
||||
invocations.ShouldContain(i => i.CorrelationId == "central-node-1-1");
|
||||
invocations.ShouldContain(i => i.CorrelationId == "central-node-2-1");
|
||||
invocations.ShouldAllBe(i => i.Ct.CanBeCanceled);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Refresh_removes_departed_nodes_and_adds_new_ones()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var current = new List<TelemetryDialTarget>
|
||||
{
|
||||
new("node-1", "http://h1:5300"),
|
||||
new("node-2", "http://h2:5300"),
|
||||
};
|
||||
|
||||
var actor = Spawn(dial, () => current.ToArray());
|
||||
dial.WaitForCount(2, Timeout);
|
||||
|
||||
var node1 = dial.Invocations.Single(i => i.Target.NodeId == "node-1");
|
||||
|
||||
// node-1 leaves, node-3 joins.
|
||||
current.Clear();
|
||||
current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
|
||||
current.Add(new TelemetryDialTarget("node-3", "http://h3:5300"));
|
||||
actor.Tell(new TelemetryDialSupervisor.RefreshNodes());
|
||||
|
||||
// node-1's dialer is cancelled and dropped.
|
||||
AwaitCondition(() => node1.Ct.IsCancellationRequested, Timeout);
|
||||
// node-3 gets a fresh dialer.
|
||||
var invocations = dial.WaitForCount(3, Timeout);
|
||||
invocations.ShouldContain(i => i.Target.NodeId == "node-3" && i.CorrelationId == "central-node-3-1");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Each_record_type_lands_on_its_sink()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var sinks = new Sinks();
|
||||
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
|
||||
|
||||
var inv = dial.WaitForCount(1, Timeout).Single();
|
||||
|
||||
inv.OnMapped(Alarm("a1"));
|
||||
inv.OnMapped(Script("s1"));
|
||||
inv.OnMapped(Health("d1"));
|
||||
inv.OnMapped(Resilience("d1"));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
sinks.Alarms.ShouldHaveSingleItem().AlarmId.ShouldBe("a1");
|
||||
sinks.Scripts.ShouldHaveSingleItem().ScriptId.ShouldBe("s1");
|
||||
sinks.Health.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
|
||||
sinks.Resilience.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
|
||||
},
|
||||
Timeout);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Stream_failure_reconnects_with_an_incremented_generation()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
|
||||
|
||||
var first = dial.WaitForCount(1, Timeout).Single();
|
||||
first.CorrelationId.ShouldBe("central-node-1-1");
|
||||
|
||||
// Transport failure → the client would end the stream after onError, so end the fake loop too.
|
||||
first.OnError(new InvalidOperationException("boom"));
|
||||
first.Complete();
|
||||
|
||||
// First retry is immediate; generation is bumped.
|
||||
var invocations = dial.WaitForCount(2, Timeout);
|
||||
invocations[1].CorrelationId.ShouldBe("central-node-1-2");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Late_event_from_a_superseded_generation_is_dropped()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var sinks = new Sinks();
|
||||
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
|
||||
|
||||
var first = dial.WaitForCount(1, Timeout).Single();
|
||||
first.OnError(new InvalidOperationException("boom"));
|
||||
first.Complete();
|
||||
dial.WaitForCount(2, Timeout); // gen-2 dialer is live.
|
||||
|
||||
// A late event arriving on the OLD (gen-1) stream must be dropped, not routed.
|
||||
first.OnMapped(Alarm("stale"));
|
||||
|
||||
// Give the message a chance to be (wrongly) routed, then assert it was not.
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
sinks.Alarms.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Late_failure_from_a_superseded_generation_does_not_trigger_a_second_reconnect()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
|
||||
|
||||
var first = dial.WaitForCount(1, Timeout).Single();
|
||||
first.OnError(new InvalidOperationException("boom"));
|
||||
first.Complete();
|
||||
dial.WaitForCount(2, Timeout); // gen-2 dialer.
|
||||
|
||||
// A duplicate/late failure from gen-1 must be ignored (no third dial).
|
||||
first.OnError(new InvalidOperationException("late duplicate"));
|
||||
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(400));
|
||||
dial.Invocations.Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Pill_flips_true_on_first_connection_and_false_only_when_all_down()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var sinks = new Sinks();
|
||||
Spawn(
|
||||
dial,
|
||||
() => new[]
|
||||
{
|
||||
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
||||
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
||||
},
|
||||
sinks);
|
||||
|
||||
var inv = dial.WaitForCount(2, Timeout);
|
||||
var node1 = inv.Single(i => i.Target.NodeId == "node-1");
|
||||
var node2 = inv.Single(i => i.Target.NodeId == "node-2");
|
||||
|
||||
// First node connects → both broadcasters flip to connected.
|
||||
node1.OnMapped(Alarm("a1"));
|
||||
AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true }), Timeout);
|
||||
|
||||
// Second node connects → still connected, no extra transition.
|
||||
node2.OnMapped(Alarm("a2"));
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
sinks.PillStates.ToArray().ShouldBe(new[] { true });
|
||||
|
||||
// node-1 drops but node-2 is still up → pill stays true.
|
||||
node1.OnError(new InvalidOperationException("down"));
|
||||
node1.Complete();
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
sinks.PillStates.ToArray().ShouldBe(new[] { true });
|
||||
|
||||
// node-2 drops too → all down → pill flips false.
|
||||
node2.OnError(new InvalidOperationException("down"));
|
||||
node2.Complete();
|
||||
AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true, false }), Timeout);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PostStop_cancels_every_dialer()
|
||||
{
|
||||
var dial = new FakeDialLoop();
|
||||
var actor = Spawn(
|
||||
dial,
|
||||
() => new[]
|
||||
{
|
||||
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
||||
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
||||
});
|
||||
|
||||
var inv = dial.WaitForCount(2, Timeout);
|
||||
|
||||
Sys.Stop(actor);
|
||||
|
||||
AwaitCondition(() => inv.All(i => i.Ct.IsCancellationRequested), Timeout);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Production_node_source_skips_null_grpc_port_and_maintenance_rows()
|
||||
{
|
||||
var factory = new InMemoryFactory(Guid.NewGuid().ToString());
|
||||
await using (var db = factory.CreateDbContext())
|
||||
{
|
||||
db.ClusterNodes.AddRange(
|
||||
Node("has-port", "h1", grpcPort: 5300, enabled: true, maintenance: false),
|
||||
Node("null-port", "h2", grpcPort: null, enabled: true, maintenance: false),
|
||||
Node("maintenance", "h3", grpcPort: 5300, enabled: true, maintenance: true),
|
||||
Node("disabled", "h4", grpcPort: 5300, enabled: false, maintenance: false));
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var source = TelemetryNodeSource.Create(factory, NullLogger.Instance);
|
||||
var targets = await source();
|
||||
|
||||
targets.ShouldHaveSingleItem();
|
||||
targets[0].NodeId.ShouldBe("has-port");
|
||||
targets[0].Endpoint.ShouldBe("http://h1:5300");
|
||||
}
|
||||
|
||||
private IActorRef Spawn(
|
||||
FakeDialLoop dial,
|
||||
Func<TelemetryDialTarget[]> nodeSource,
|
||||
Sinks? sinks = null)
|
||||
{
|
||||
sinks ??= new Sinks();
|
||||
var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600 };
|
||||
return Sys.ActorOf(TelemetryDialSupervisor.Props(
|
||||
() => Task.FromResult<IReadOnlyList<TelemetryDialTarget>>(nodeSource()),
|
||||
dial.Loop,
|
||||
sinks.AlarmBroadcaster,
|
||||
sinks.ScriptBroadcaster,
|
||||
sinks.HealthStore,
|
||||
sinks.ResilienceStore,
|
||||
options));
|
||||
}
|
||||
|
||||
private static AlarmTransitionEvent Alarm(string id) =>
|
||||
new(id, "Area/Line/Eq", "hi", "Activated", 500, "msg", "system", DateTime.UtcNow);
|
||||
|
||||
private static ScriptLogEntry Script(string id) =>
|
||||
new(id, "Info", "msg", DateTime.UtcNow, null, null, null);
|
||||
|
||||
private static DriverHealthChanged Health(string id) =>
|
||||
new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow);
|
||||
|
||||
private static DriverResilienceStatusChanged Resilience(string id) =>
|
||||
new(id, "host", false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow);
|
||||
|
||||
private static ClusterNode Node(string id, string host, int? grpcPort, bool enabled, bool maintenance) =>
|
||||
new()
|
||||
{
|
||||
NodeId = id,
|
||||
ClusterId = "C1",
|
||||
Host = host,
|
||||
GrpcPort = grpcPort,
|
||||
Enabled = enabled,
|
||||
MaintenanceMode = maintenance,
|
||||
ApplicationUri = $"urn:{id}",
|
||||
CreatedBy = "test",
|
||||
};
|
||||
|
||||
/// <summary>A captured dial-loop invocation with the callbacks the test drives by hand.</summary>
|
||||
private sealed class DialInvocation
|
||||
{
|
||||
private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
public required TelemetryDialTarget Target { get; init; }
|
||||
public required string CorrelationId { get; init; }
|
||||
public required Action<object> OnMapped { get; init; }
|
||||
public required Action<Exception> OnError { get; init; }
|
||||
public required CancellationToken Ct { get; init; }
|
||||
|
||||
public Task Completion => _completion.Task;
|
||||
|
||||
/// <summary>Ends the fake loop (as the client returns after a server-ended or post-onError stream).</summary>
|
||||
public void Complete() => _completion.TrySetResult();
|
||||
}
|
||||
|
||||
/// <summary>A fake <see cref="TelemetryDialLoop"/> that records each invocation and blocks until told to end.</summary>
|
||||
private sealed class FakeDialLoop
|
||||
{
|
||||
private readonly ConcurrentQueue<DialInvocation> _invocations = new();
|
||||
|
||||
public IReadOnlyList<DialInvocation> Invocations => _invocations.ToArray();
|
||||
|
||||
public TelemetryDialLoop Loop => async (target, correlationId, onMapped, onError, ct) =>
|
||||
{
|
||||
var invocation = new DialInvocation
|
||||
{
|
||||
Target = target,
|
||||
CorrelationId = correlationId,
|
||||
OnMapped = onMapped,
|
||||
OnError = onError,
|
||||
Ct = ct,
|
||||
};
|
||||
_invocations.Enqueue(invocation);
|
||||
|
||||
// Complete when the test ends the loop OR the supervisor cancels us.
|
||||
await using var reg = ct.Register(() => invocation.Complete());
|
||||
await invocation.Completion;
|
||||
};
|
||||
|
||||
public IReadOnlyList<DialInvocation> WaitForCount(int count, TimeSpan timeout)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + timeout;
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
var snapshot = _invocations.ToArray();
|
||||
if (snapshot.Length >= count)
|
||||
{
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
Thread.Sleep(10);
|
||||
}
|
||||
|
||||
throw new Xunit.Sdk.XunitException(
|
||||
$"Expected at least {count} dial invocations within {timeout}, saw {_invocations.Count}.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The four fake sinks plus the pill-transition record.</summary>
|
||||
private sealed class Sinks
|
||||
{
|
||||
public RecordingBroadcaster<AlarmTransitionEvent> AlarmBroadcaster { get; }
|
||||
public RecordingBroadcaster<ScriptLogEntry> ScriptBroadcaster { get; }
|
||||
public RecordingHealthStore HealthStore { get; } = new();
|
||||
public RecordingResilienceStore ResilienceStore { get; } = new();
|
||||
|
||||
public ConcurrentQueue<AlarmTransitionEvent> Alarms { get; } = new();
|
||||
public ConcurrentQueue<ScriptLogEntry> Scripts { get; } = new();
|
||||
public ConcurrentQueue<bool> PillStates { get; } = new();
|
||||
|
||||
public ConcurrentQueue<DriverHealthChanged> Health => HealthStore.Upserts;
|
||||
public ConcurrentQueue<DriverResilienceStatusChanged> Resilience => ResilienceStore.Upserts;
|
||||
|
||||
public Sinks()
|
||||
{
|
||||
// Both broadcasters share the ONE pill-transition record so a test can assert the alarm and
|
||||
// script pills move together (matching the supervisor calling SetConnected on both).
|
||||
AlarmBroadcaster = new RecordingBroadcaster<AlarmTransitionEvent>(Alarms, PillStates);
|
||||
ScriptBroadcaster = new RecordingBroadcaster<ScriptLogEntry>(Scripts, null);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class RecordingBroadcaster<T>(ConcurrentQueue<T> published, ConcurrentQueue<bool>? pill)
|
||||
: IInProcessBroadcaster<T>
|
||||
{
|
||||
public event Action<T>? Received;
|
||||
public event Action<bool>? ConnectionStateChanged;
|
||||
|
||||
public bool IsConnected { get; private set; }
|
||||
|
||||
public void Publish(T item)
|
||||
{
|
||||
published.Enqueue(item);
|
||||
Received?.Invoke(item);
|
||||
}
|
||||
|
||||
public void SetConnected(bool connected)
|
||||
{
|
||||
IsConnected = connected;
|
||||
pill?.Enqueue(connected);
|
||||
ConnectionStateChanged?.Invoke(connected);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class RecordingHealthStore : IDriverStatusSnapshotStore
|
||||
{
|
||||
public ConcurrentQueue<DriverHealthChanged> Upserts { get; } = new();
|
||||
|
||||
public event Action<DriverHealthChanged>? SnapshotChanged;
|
||||
|
||||
public void Upsert(DriverHealthChanged snapshot)
|
||||
{
|
||||
Upserts.Enqueue(snapshot);
|
||||
SnapshotChanged?.Invoke(snapshot);
|
||||
}
|
||||
|
||||
public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot)
|
||||
{
|
||||
snapshot = null!;
|
||||
return false;
|
||||
}
|
||||
|
||||
public IReadOnlyCollection<DriverHealthChanged> GetAll() => Upserts.ToArray();
|
||||
}
|
||||
|
||||
private sealed class RecordingResilienceStore : IDriverResilienceStatusStore
|
||||
{
|
||||
public ConcurrentQueue<DriverResilienceStatusChanged> Upserts { get; } = new();
|
||||
|
||||
public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
|
||||
|
||||
public void Upsert(DriverResilienceStatusChanged snapshot)
|
||||
{
|
||||
Upserts.Enqueue(snapshot);
|
||||
SnapshotChanged?.Invoke(snapshot);
|
||||
}
|
||||
|
||||
public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) =>
|
||||
Upserts.ToArray();
|
||||
|
||||
public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => Upserts.ToArray();
|
||||
}
|
||||
|
||||
private sealed class InMemoryFactory(string name) : IDbContextFactory<OtOpcUaConfigDbContext>
|
||||
{
|
||||
public OtOpcUaConfigDbContext CreateDbContext() =>
|
||||
new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
|
||||
.UseInMemoryDatabase(name)
|
||||
.Options);
|
||||
|
||||
public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(CreateDbContext());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user