feat(hosts): driver diagnostics channel on /hosts, S7 negotiated PDU size first (#640)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 16:57:38 -04:00
parent a8b07ae12d
commit c2bdb7d5de
23 changed files with 469 additions and 20 deletions
+3 -2
View File
@@ -1113,8 +1113,9 @@ host logger factory (wired in `DriverFactoryBootstrap`; the FOCAS and TwinCAT fa
- **S7 (#632/#638):** STRING/WSTRING/CHAR/WCHAR and DTL/DATE/TOD/TIME/S5TIME codecs (`S7ValueCodecs`; TOD/TIME/S5TIME
surface as Int32 ms); an undecodable buffer or unrepresentable write is `BadOutOfRange` with no health impact.
Optional `localTsap`/`remoteTsap`/`connectionType` override the slot TSAP (`remoteTsap` ⊕ `connectionType`). Test
Connect reads 2 bytes of `MB0` so a PUT/GET-disabled CPU fails the probe. The negotiated PDU size is logged and on
`S7Driver.GetDiagnostics()` only — there is no driver-diagnostics surface on `/hosts`.
Connect reads 2 bytes of `MB0` so a PUT/GET-disabled CPU fails the probe. The negotiated PDU size is logged, on
`S7Driver.GetDiagnostics()`, and (#640) on `/hosts` via the optional `IDriverDiagnostics` bag (string map on
`DriverHealthChanged.Diagnostics` / proto `has_diagnostics`+`diagnostics`; its digest is in the publish fingerprint).
- **TwinCAT (#637):** TIME/TOD publish Float64 ms, DATE/DT a UTC DateTime; a tag authored `UDInt` keeps the raw counter
(and keeps historizing — DateTime is not historized).
- **AbLegacy (#636):** `AbLegacyAddress` (in `AbLegacy.Contracts`) is family-aware: PLC-5 `I:`/`O:` indices are OCTAL
+17
View File
@@ -112,6 +112,23 @@ it — but it does mean the panel cannot answer "which node saw this?". Adding t
widening the wire record, not changing the page. The one sub-instance breakdown the channel *does*
carry is per-host connectivity, below.
### Driver diagnostics ride `driver-health` (Gitea #640)
A driver may implement the optional `IDriverDiagnostics` (Core.Abstractions): a pure in-memory
`IReadOnlyDictionary<string,string> GetDiagnostics()` of operator-facing facts that are not driver
*state* (S7's negotiated PDU size is the first; others opt in later). `DriverInstanceActor` pulls it on
every health publish (try/caught, unguarded by the invoker like `GetHostStatuses`), bounds it
(`DriverDiagnostics.Bound`: 16 entries, text truncated to 200 chars, ordinal key order), and carries it as
`DriverHealthChanged.Diagnostics` -> proto `DriverHealth.has_diagnostics = 13` / `map<string,string>
diagnostics = 14` (additive; old nodes never set them, old central ignores them) -> `HostsDriverRow.DiagnosticItems`,
rendered as a small key/value list under the driver name on `/hosts` (nothing when empty).
- **null ≠ empty**: null = driver lacks the capability; the presence flag preserves an empty bag.
- **The diagnostics digest is part of the publish fingerprint, and must stay there** - a diagnostic fact
(e.g. PDU size changing on a reconnect) moves no other component, so without it the dedup swallows the
publish. Pinned by `DriverInstanceActorDiagnosticsTests.A_diagnostics_change_republishes_despite_otherwise_unchanged_health`
(stable-health stub - the shared `StubDriver` returns `UtcNow` and would pass vacuously).
### Per-host connectivity rides `driver-health` (Gitea #521)
`DriverHealthChanged.HostStatuses` carries each driver's `IHostConnectivityProbe.GetHostStatuses()`
+4 -2
View File
@@ -223,8 +223,10 @@ The PDU size the CPU negotiates during the connection handshake (typically 240 b
extended PDU) is snapshotted at every (re)connect and exposed as `S7Driver.NegotiatedPduSize` /
`S7Driver.GetDiagnostics()["S7.NegotiatedPduSize"]` (0 before the first connect and after shutdown). It is
also written to the `S7Driver connected` / `reconnected` log lines and to the AdminUI Test Connect
message (`S7 connected (CPU S71500, PDU 240 bytes)`). `DriverHealth` has no diagnostics bag in this
codebase, so the value is not yet on the `/hosts` health snapshot — adding one is a Core change.
message (`S7 connected (CPU S71500, PDU 240 bytes)`). S7 also implements the optional
`IDriverDiagnostics` capability (explicitly, string-valued: `S7.NegotiatedPduSize` = `"480"`, omitted while 0 =
not negotiated yet), so the value rides the driver-health snapshot and shows under the driver name on the
AdminUI `/hosts` page (Gitea #640). See `docs/Telemetry.md` ("Driver diagnostics ride `driver-health`").
## Capability surface
@@ -59,6 +59,12 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
/// <para><b>Value-equality caveat</b> — the same one <paramref name="HostStatuses"/> carries: the
/// record's generated <c>Equals</c> compares the sample list by REFERENCE.</para>
/// </param>
/// <param name="Diagnostics">
/// Operator-facing driver diagnostics from <c>IDriverDiagnostics.GetDiagnostics()</c> (e.g. S7's
/// negotiated PDU size), bounded and in ordinal key order; null when the driver does not implement the
/// capability (Gitea #640). Additive, never a state change. Same by-reference value-equality caveat as
/// <paramref name="HostStatuses"/>.
/// </param>
public sealed record DriverHealthChanged(
string ClusterId,
string DriverInstanceId,
@@ -70,7 +76,8 @@ public sealed record DriverHealthChanged(
DateTime? RediscoveryNeededUtc = null,
string? RediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? HostStatuses = null,
SkippedTagReport? SkippedTags = null)
SkippedTagReport? SkippedTags = null,
IReadOnlyDictionary<string, string>? Diagnostics = null)
{
/// <summary>
/// DPS topic name. Both the runtime <c>AkkaDriverHealthPublisher</c> and the AdminUI
@@ -82,6 +82,12 @@ message DriverHealth {
// proto3, which a repeated field does not. So no companion has_ flag (and one would collide: protoc
// already generates a HasSkippedTags property for this field).
SkippedTags skipped_tags = 12;
// Driver diagnostics bag (Gitea #640), e.g. S7's negotiated PDU size. Additive: an old node never sets
// these and an old central ignores them. A map cannot distinguish absent from empty, and the two differ
// ("driver has no IDriverDiagnostics" vs "it has one and currently has nothing to say"), so the presence
// flag carries that bit - the same shape as has_host_statuses.
bool has_diagnostics = 13;
map<string, string> diagnostics = 14;
}
// Mirrors ZB.MOM.WW.OtOpcUa.Core.Abstractions.SkippedTagReport.
@@ -0,0 +1,58 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// <summary>
/// Optional driver capability: expose a small bag of operator-facing diagnostic key/value pairs
/// (e.g. S7's negotiated PDU size) that ride the driver-health snapshot to the AdminUI <c>/hosts</c>
/// page (Gitea #640).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why a bag rather than fields on <see cref="DriverHealth"/>.</b> The facts worth showing are
/// driver-specific and do not change the driver's <i>state</i>; a typed field per driver would
/// widen the core record and the telemetry proto for every driver. Same additive-never-a-state-change
/// rule as <see cref="ISkippedTagReporter"/>.
/// </para>
/// <para>
/// Values are strings so the telemetry proto stays a plain <c>map&lt;string,string&gt;</c>. Keep the
/// bag small (<see cref="DriverDiagnostics.MaxEntries"/> entries, text truncated to
/// <see cref="DriverDiagnostics.MaxValueLength"/>): it is published on every state change.
/// </para>
/// </remarks>
public interface IDriverDiagnostics
{
/// <summary>
/// Snapshot of the driver's diagnostics. Must be a pure in-memory read with no I/O — it is called on
/// the health-publish path, deliberately NOT through the resilience pipeline.
/// </summary>
/// <returns>The key/value pairs; empty when there is nothing to show yet.</returns>
IReadOnlyDictionary<string, string> GetDiagnostics();
}
/// <summary>Bounds and helpers shared by every consumer of <see cref="IDriverDiagnostics"/>.</summary>
public static class DriverDiagnostics
{
/// <summary>Maximum number of entries carried on a health snapshot; extras are dropped (ordinal key order).</summary>
public const int MaxEntries = 16;
/// <summary>Maximum length of a key or value carried on a health snapshot; longer text is truncated.</summary>
public const int MaxValueLength = 200;
/// <summary>
/// Normalises a driver's bag for publishing: ordinal key order, capped entry count, truncated text.
/// Deterministic so the same facts always produce the same fingerprint digest.
/// </summary>
/// <param name="raw">The bag the driver returned.</param>
/// <returns>A bounded, ordered copy.</returns>
public static IReadOnlyDictionary<string, string> Bound(IReadOnlyDictionary<string, string> raw)
{
var result = new SortedDictionary<string, string>(StringComparer.Ordinal);
foreach (var kv in raw.OrderBy(k => k.Key, StringComparer.Ordinal))
{
if (result.Count >= MaxEntries) break;
result[Truncate(kv.Key ?? "")] = Truncate(kv.Value ?? "");
}
return result;
}
private static string Truncate(string s) => s.Length <= MaxValueLength ? s : s[..MaxValueLength];
}
@@ -34,6 +34,10 @@ public interface IDriverHealthPublisher
/// <see cref="ISkippedTagReporter"/>. Contradicts the "Healthy, 0 errors, reads advancing" reading
/// of a driver that is in fact ignoring part or all of its configuration (Gitea #541).
/// </param>
/// <param name="diagnostics">
/// Bounded driver diagnostics from <see cref="IDriverDiagnostics.GetDiagnostics"/>; null when the
/// driver is not an <see cref="IDriverDiagnostics"/> (Gitea #640).
/// </param>
void Publish(
string clusterId,
string driverInstanceId,
@@ -42,7 +46,8 @@ public interface IDriverHealthPublisher
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null);
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null);
}
/// <summary>
@@ -65,6 +70,7 @@ public sealed class NullDriverHealthPublisher : IDriverHealthPublisher
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{ /* no-op */ }
}
@@ -32,7 +32,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7;
/// </para>
/// </remarks>
public sealed class S7Driver
: IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDisposable, IAsyncDisposable
: IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDriverDiagnostics, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table (Gitea #541).</summary>
private readonly SkippedTagLedger _skippedTags = new();
@@ -463,6 +463,19 @@ public sealed class S7Driver
["S7.NegotiatedPduSize"] = NegotiatedPduSize,
};
/// <summary>
/// <see cref="IDriverDiagnostics"/> view of the same facts, as strings, for the <c>/hosts</c> page
/// (Gitea #640). Empty before the first connect (PDU size 0 is "not negotiated yet", not a size).
/// </summary>
/// <returns>The diagnostics bag.</returns>
IReadOnlyDictionary<string, string> IDriverDiagnostics.GetDiagnostics()
{
var pdu = NegotiatedPduSize;
return pdu > 0
? new Dictionary<string, string> { ["S7.NegotiatedPduSize"] = pdu.ToString(System.Globalization.CultureInfo.InvariantCulture) }
: new Dictionary<string, string>();
}
private static readonly HashSet<S7DataType> ByteAnchoredDataTypes = new()
{
S7DataType.Int64,
@@ -189,6 +189,17 @@ else
@* Own line — inline after the name the two identifiers ran together. *@
<div class="mono small text-muted">@d.DriverInstanceId</div>
}
@if (d.DiagnosticItems.Count > 0)
{
@* Driver-reported diagnostics (Gitea #640), e.g. S7 negotiated PDU size.
Renders nothing for a driver with none. *@
<div class="small text-muted" data-testid="driver-diagnostics">
@foreach (var kv in d.DiagnosticItems)
{
<div class="mono">@kv.Key: @kv.Value</div>
}
</div>
}
</td>
<td>@(d.DriverType ?? "—")</td>
<td><span class="chip @DriverChipClass(d.State)">@d.State</span></td>
@@ -52,6 +52,8 @@ public sealed record HostsDriverInstanceInfo(
/// is no answer — the driver cannot report skips, or it faulted before building a tag table. Lets
/// "ignoring 20 of 22 tags" show even while every other column on the row — state, last read, error
/// count, last error — reads perfectly healthy (Gitea #541).</param>
/// <param name="Diagnostics">Driver-reported diagnostics (e.g. S7 negotiated PDU size); null or empty when
/// the driver has none to show (Gitea #640).</param>
/// <param name="Enabled">Whether the matching configured instance is authored enabled; <see langword="true"/>
/// when no instance matched. Fed to the row's expandable <c>DriverStatusPanel</c> (Gitea #546). An
/// unmatched row defaults to enabled on purpose: the row only exists because a live snapshot arrived,
@@ -62,8 +64,16 @@ public sealed record HostsDriverRow(
DateTime? RediscoveryNeededUtc = null, string? RediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? HostStatuses = null,
SkippedTagReport? SkippedTags = null,
bool Enabled = true)
bool Enabled = true,
IReadOnlyDictionary<string, string>? Diagnostics = null)
{
/// <summary>Diagnostics as ordered key/value pairs for rendering; empty when there are none, in which
/// case the page renders nothing.</summary>
public IReadOnlyList<KeyValuePair<string, string>> DiagnosticItems { get; } =
(Diagnostics ?? new Dictionary<string, string>())
.OrderBy(k => k.Key, StringComparer.Ordinal)
.ToList();
/// <summary>Whether this driver is ignoring at least one authored tag — the condition the whole
/// column exists to make visible.</summary>
public bool HasSkippedTags => SkippedTags is { SkippedCount: > 0 };
@@ -177,7 +187,8 @@ public static class HostsDriverView
s.RediscoveryReason,
s.HostStatuses,
s.SkippedTags,
instance?.Enabled ?? true);
instance?.Enabled ?? true,
s.Diagnostics);
})
.OrderBy(d => d.Name ?? d.DriverInstanceId, StringComparer.OrdinalIgnoreCase)
.ThenBy(d => d.DriverInstanceId, StringComparer.OrdinalIgnoreCase)
@@ -130,7 +130,8 @@ public static class TelemetryProtoMapCentral
RediscoveryNeededUtc: msg.RediscoveryNeededUtc?.ToDateTime(),
RediscoveryReason: msg.HasRediscoveryReason ? msg.RediscoveryReason : null,
HostStatuses: ToHostStatuses(msg),
SkippedTags: ToSkippedTags(msg));
SkippedTags: ToSkippedTags(msg),
Diagnostics: msg.HasDiagnostics ? new Dictionary<string, string>(msg.Diagnostics) : null);
}
/// <summary>
@@ -164,6 +164,14 @@ public static class TelemetryProtoMapNode
msg.SkippedTags = payload;
}
// Gitea #640. Presence flag first - an empty map cannot say whether the driver has the capability.
if (e.Diagnostics is not null)
{
msg.HasDiagnostics = true;
foreach (var kv in e.Diagnostics)
msg.Diagnostics[kv.Key ?? ""] = kv.Value ?? "";
}
return msg;
}
@@ -38,7 +38,8 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{
var msg = new DriverHealthChanged(
clusterId,
@@ -51,7 +52,8 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher
rediscoveryNeededUtc,
rediscoveryReason,
hostStatuses,
skippedTags);
skippedTags,
diagnostics);
DistributedPubSub.Get(_system).Mediator.Tell(new Publish(TopicName, msg));
// Phase 5: fan the same snapshot into the node-local live-telemetry hub (no-op until a gRPC
// client subscribes). The DPS publish above is unchanged — the hub is a strictly additive tap.
@@ -1153,6 +1153,28 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
}
}
/// <summary>Bounded driver diagnostics, or null when the driver is not an <see cref="IDriverDiagnostics"/>
/// (Gitea #640). Pulled fresh and called UNGUARDED for the same reasons as
/// <see cref="CurrentSkippedTags"/>; a throwing implementation loses only the diagnostics.</summary>
private IReadOnlyDictionary<string, string>? CurrentDiagnostics()
{
if (_driver is not IDriverDiagnostics diag) return null;
try
{
return DriverDiagnostics.Bound(diag.GetDiagnostics());
}
catch (Exception ex)
{
_log.Warning(ex, "DriverInstance {Id}: GetDiagnostics threw during health publish; omitting diagnostics", _driverInstanceId);
return null;
}
}
/// <summary>Value digest of the diagnostics bag for the publish fingerprint. Null in, null out. The bag is
/// already in ordinal key order (<see cref="DriverDiagnostics.Bound"/>), so equal facts digest equally.</summary>
private static string? DiagnosticsDigest(IReadOnlyDictionary<string, string>? d) =>
d is null ? null : string.Join('|', d.Select(kv => $"{kv.Key}={kv.Value}"));
/// <summary>Records the driver's rediscovery raise and re-publishes health so the signal reaches the
/// AdminUI promptly rather than waiting for the next 30 s heartbeat.
/// <para><b>Advisory only.</b> The served address space is deliberately NOT rebuilt: v3 authors raw tags
@@ -1357,6 +1379,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
var errorCount = ErrorCount5Min();
var hostStatuses = CurrentHostStatuses();
var skippedTags = CurrentSkippedTags();
var diagnostics = CurrentDiagnostics();
// _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an
// otherwise-unchanged Healthy driver leaves (state, lastSuccess, lastError, errorCount)
// identical, so without it the dedup below would swallow the very publish that carries the
@@ -1375,14 +1398,20 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// is identical to a clean driver's and the publish that carries the signal is exactly the one
// dedup would swallow. Flattened to a STRING for the same reason as the host digest — the
// report holds an IReadOnlyList, which a tuple compares by reference.
//
// The diagnostics digest (Gitea #640) is the fourth instance: a diagnostic fact (S7's PDU size
// changing across a reconnect) moves none of the other components, so without it the publish
// that carries it is deduped away. Flattened to a STRING - the bag is a dictionary, which a
// tuple compares by reference.
var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount,
_rediscoveryNeededUtc, HostStatusDigest(hostStatuses), SkippedTagDigest(skippedTags));
_rediscoveryNeededUtc, HostStatusDigest(hostStatuses), SkippedTagDigest(skippedTags),
DiagnosticsDigest(diagnostics));
if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint))
return observed;
_lastPublishedFingerprint = fingerprint;
_healthPublisher.Publish(
_clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason,
hostStatuses, skippedTags);
hostStatuses, skippedTags, diagnostics);
}
catch (Exception ex)
{
@@ -1418,7 +1447,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <summary>Fingerprint of the last <see cref="PublishHealthSnapshot"/> call; null until first publish.</summary>
private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount,
DateTime? RediscoveryNeededUtc, string? HostStatusDigest, string? SkippedTagDigest)? _lastPublishedFingerprint;
DateTime? RediscoveryNeededUtc, string? HostStatusDigest, string? SkippedTagDigest,
string? DiagnosticsDigest)? _lastPublishedFingerprint;
/// <inheritdoc />
protected override void PostStop()
@@ -140,13 +140,16 @@ public sealed class S7TsapAndPduTests
using var drv = new S7Driver(opts, "s7-pdu", factory);
drv.NegotiatedPduSize.ShouldBe(0);
drv.GetDiagnostics()["S7.NegotiatedPduSize"].ShouldBe(0);
((IDriverDiagnostics)drv).GetDiagnostics().ShouldBeEmpty("0 is 'not negotiated yet', not a size (#640)");
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
drv.NegotiatedPduSize.ShouldBe(480);
drv.GetDiagnostics()["S7.NegotiatedPduSize"].ShouldBe(480);
((IDriverDiagnostics)drv).GetDiagnostics()["S7.NegotiatedPduSize"].ShouldBe("480");
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
drv.NegotiatedPduSize.ShouldBe(0);
((IDriverDiagnostics)drv).GetDiagnostics().ShouldBeEmpty();
}
[Fact]
@@ -272,6 +272,24 @@ public sealed class HostsDriverViewTests
row.SkippedTags!.SkippedCount.ShouldBe(20);
}
/// <summary>#640: diagnostics ride the snapshot onto the row, ordered by key; none renders nothing.</summary>
[Fact]
public void Diagnostics_reach_the_row_ordered_and_absent_diagnostics_yield_no_items()
{
var with = Snap("MAIN", "s7-a") with
{
Diagnostics = new Dictionary<string, string> { ["Z"] = "2", ["S7.NegotiatedPduSize"] = "480" },
};
var without = Snap("MAIN", "s7-b");
var rows = HostsDriverView.Build([with, without], nodes: null, instances: null).Single().Drivers;
var a = rows.Single(r => r.DriverInstanceId == "s7-a");
a.DiagnosticItems.Select(k => k.Key).ShouldBe(["S7.NegotiatedPduSize", "Z"]);
a.DiagnosticItems[0].Value.ShouldBe("480");
rows.Single(r => r.DriverInstanceId == "s7-b").DiagnosticItems.ShouldBeEmpty();
}
/// <summary>
/// A truncated sample must SAY it is truncated. Without the note a capped list reads as the whole
/// list, and an operator who fixes the twenty named tags believes the job is done.
@@ -276,6 +276,43 @@ public sealed class TelemetryProtoMapCentralTests
TelemetryProtoMapCentral.ToHealth(absentSkips).SkippedTags.ShouldBeNull();
}
/// <summary>
/// #640: the diagnostics bag round-trips with the null vs empty distinction intact, and a pre-#640
/// node (field never set) decodes to null.
/// </summary>
[Fact]
public void ToHealth_diagnostics_round_trip_with_the_null_vs_empty_distinction_intact()
{
var populated = new DriverHealth
{
ClusterId = "c1", DriverInstanceId = "d1", State = "Healthy",
PublishedUtc = Timestamp.FromDateTime(SampleUtc),
HasDiagnostics = true,
Diagnostics = { ["S7.NegotiatedPduSize"] = "480", ["Other"] = "x" },
};
var mapped = TelemetryProtoMapCentral.ToHealth(populated).Diagnostics;
mapped.ShouldNotBeNull();
mapped!.Count.ShouldBe(2);
mapped["S7.NegotiatedPduSize"].ShouldBe("480");
var empty = new DriverHealth
{
ClusterId = "c1", DriverInstanceId = "d1", State = "Healthy",
PublishedUtc = Timestamp.FromDateTime(SampleUtc),
HasDiagnostics = true,
};
var mappedEmpty = TelemetryProtoMapCentral.ToHealth(empty).Diagnostics;
mappedEmpty.ShouldNotBeNull();
mappedEmpty!.ShouldBeEmpty();
var absent = new DriverHealth
{
ClusterId = "c1", DriverInstanceId = "d1", State = "Healthy",
PublishedUtc = Timestamp.FromDateTime(SampleUtc),
};
TelemetryProtoMapCentral.ToHealth(absent).Diagnostics.ShouldBeNull();
}
/// <summary>
/// An unparseable host state degrades to <see cref="HostState.Unknown"/> rather than throwing. A node
/// running a newer build that added an enum member must not be able to kill central's telemetry
@@ -248,6 +248,33 @@ public sealed class TelemetryStreamGrpcServiceTests
.DriverHealth.HasHostStatuses.ShouldBeFalse();
}
/// <summary>
/// Node side of the #640 diagnostics carry: presence flag + entries written; a capability-less driver
/// (null) leaves the flag clear, and an empty bag still sets it.
/// </summary>
[Fact]
public void ToProto_health_writes_the_diagnostics_presence_flag_and_entries()
{
var withDiag = new DriverHealthChanged(
"cluster-a", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow,
Diagnostics: new Dictionary<string, string> { ["S7.NegotiatedPduSize"] = "480" });
var evt = TelemetryProtoMapNode.ToProto(new TelemetryItem.Health(withDiag), "c");
evt.DriverHealth.HasDiagnostics.ShouldBeTrue();
evt.DriverHealth.Diagnostics["S7.NegotiatedPduSize"].ShouldBe("480");
var empty = new DriverHealthChanged(
"cluster-a", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow,
Diagnostics: new Dictionary<string, string>());
var emptyEvt = TelemetryProtoMapNode.ToProto(new TelemetryItem.Health(empty), "c");
emptyEvt.DriverHealth.HasDiagnostics.ShouldBeTrue();
emptyEvt.DriverHealth.Diagnostics.ShouldBeEmpty();
var none = new DriverHealthChanged("cluster-a", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow);
var noneEvt = TelemetryProtoMapNode.ToProto(new TelemetryItem.Health(none), "c");
noneEvt.DriverHealth.HasDiagnostics.ShouldBeFalse();
noneEvt.DriverHealth.Diagnostics.ShouldBeEmpty();
}
[Fact]
public async Task Client_disconnect_mid_stream_ends_cleanly_without_leaking_a_slot()
{
@@ -0,0 +1,187 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Covers the <see cref="IDriverDiagnostics"/> consumer (Gitea #640). A diagnostic fact (S7's negotiated
/// PDU size) moves no other fingerprint component, so unless its digest is in
/// <c>PublishHealthSnapshot</c>'s fingerprint the dedup swallows the very publish that carries it.
/// <para>The stub's health is deliberately STABLE: the shared <c>StubDriver.GetHealth()</c> returns
/// <c>UtcNow</c>, so the dedup never engages and a test on it passes vacuously.</para>
/// </summary>
[Trait("Category", "Unit")]
public sealed class DriverInstanceActorDiagnosticsTests : RuntimeActorTestBase
{
/// <summary>Base case: the bag reaches the publisher.</summary>
[Fact]
public void Diagnostics_reach_the_health_publisher()
{
var driver = new DiagStubDriver();
driver.Set(("S7.NegotiatedPduSize", "480"));
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(driver, publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() =>
{
var latest = publisher.Published.LastOrDefault();
latest.ShouldNotBeNull();
latest!["S7.NegotiatedPduSize"].ShouldBe("480");
}, TimeSpan.FromSeconds(3));
}
/// <summary>
/// The load-bearing case. Falsifiability: publish count must STRICTLY increase across the change;
/// drop <c>DiagnosticsDigest</c> from the fingerprint and this goes red.
/// </summary>
[Fact]
public void A_diagnostics_change_republishes_despite_otherwise_unchanged_health()
{
var driver = new DiagStubDriver();
driver.Set(("S7.NegotiatedPduSize", "240"));
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(driver, publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
var before = publisher.Published.Count;
driver.Set(("S7.NegotiatedPduSize", "480"));
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(before), TimeSpan.FromSeconds(3));
publisher.Published[^1]!["S7.NegotiatedPduSize"].ShouldBe("480");
}
/// <summary>Equal-but-distinct bags must be deduped: the digest compares by VALUE, not reference.</summary>
[Fact]
public void An_unchanged_diagnostics_bag_does_not_defeat_the_dedup()
{
var driver = new DiagStubDriver();
driver.Set(("S7.NegotiatedPduSize", "480"));
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(driver, publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
var before = publisher.Published.Count;
driver.Set(("S7.NegotiatedPduSize", "480"));
ExpectNoMsg(TimeSpan.FromMilliseconds(600));
publisher.Published.Count.ShouldBe(before);
}
/// <summary>A driver without the capability publishes null, not an empty bag.</summary>
[Fact]
public void A_driver_without_the_capability_publishes_null()
{
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(new StubDriver(), publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
publisher.Published.ShouldAllBe(p => p == null);
}
/// <summary>A throwing implementation loses only the diagnostics, not the health publish.</summary>
[Fact]
public void A_throwing_diagnostics_provider_degrades_to_null_without_killing_the_health_publish()
{
var driver = new DiagStubDriver { Throw = true };
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(driver, publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
publisher.Published[^1].ShouldBeNull();
}
/// <summary>The bag is capped and truncated before it is published.</summary>
[Fact]
public void Oversized_bags_are_bounded()
{
var driver = new DiagStubDriver();
driver.Set([.. Enumerable.Range(0, 40).Select(i => ($"k{i:D2}", new string('x', 500)))]);
var publisher = new RecordingHealthPublisher();
var actor = SpawnDriverActor(driver, publisher);
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitAssert(() =>
{
var latest = publisher.Published.LastOrDefault();
latest.ShouldNotBeNull();
latest!.Count.ShouldBe(DriverDiagnostics.MaxEntries);
latest.Values.ShouldAllBe(v => v.Length == DriverDiagnostics.MaxValueLength);
}, TimeSpan.FromSeconds(3));
}
private IActorRef SpawnDriverActor(IDriver driver, IDriverHealthPublisher publisher)
{
var parent = CreateTestProbe();
parent.IgnoreMessages(_ => true);
return parent.ChildActorOf(DriverInstanceActor.Props(
driver, healthPublisher: publisher, healthPollInterval: TimeSpan.FromMilliseconds(50)));
}
private sealed class RecordingHealthPublisher : IDriverHealthPublisher
{
private readonly List<IReadOnlyDictionary<string, string>?> _published = [];
public IReadOnlyList<IReadOnlyDictionary<string, string>?> Published
{
get { lock (_published) return _published.ToArray(); }
}
public void Publish(
string clusterId,
string driverInstanceId,
DriverHealth health,
int errorCount5Min,
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{
lock (_published) _published.Add(diagnostics);
}
}
/// <summary>Stable-health driver exposing <see cref="IDriverDiagnostics"/>.</summary>
private sealed class DiagStubDriver : IDriver, IDriverDiagnostics
{
private static readonly DateTime FixedLastRead = new(2026, 10, 8, 12, 0, 0, DateTimeKind.Utc);
private volatile IReadOnlyDictionary<string, string> _bag = new Dictionary<string, string>();
public bool Throw { get; init; }
public string DriverInstanceId => "diag-stub-1";
public string DriverType => "Stub";
public void Set(params (string Key, string Value)[] items) =>
_bag = items.ToDictionary(i => i.Key, i => i.Value);
public IReadOnlyDictionary<string, string> GetDiagnostics() =>
Throw ? throw new InvalidOperationException("broken") : _bag;
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
public DriverHealth GetHealth() => new(DriverState.Healthy, FixedLastRead, null);
}
}
@@ -211,7 +211,8 @@ public sealed class DriverInstanceActorHostStatusTests : RuntimeActorTestBase
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{
lock (_published) _published.Add(new HealthPublish(health, hostStatuses));
}
@@ -187,7 +187,8 @@ public sealed class DriverInstanceActorRediscoverySignalTests : RuntimeActorTest
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{
lock (_published)
{
@@ -209,7 +209,8 @@ public sealed class DriverInstanceActorSkippedTagTests : RuntimeActorTestBase
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
{
lock (_published) _published.Add(new HealthPublish(health, skippedTags));
}
@@ -124,7 +124,8 @@ public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActor
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null)
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
=> Interlocked.Increment(ref _count);
}
}