diff --git a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json index b9e419f2..6acb1ceb 100644 --- a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json +++ b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json @@ -5,7 +5,7 @@ { "id": "T1", "subject": "Extend AddressSpaceApplyOutcome with RebuildFailed/FailedNodes + SafeRebuild returns bool (01/S-1)", "status": "completed", "blockedBy": [] }, { "id": "T2", "subject": "Safe* helpers return bool; Apply tallies removal-pass failures into FailedNodes", "status": "completed", "blockedBy": ["T1"] }, { "id": "T3", "subject": "Materialise* passes return swallowed-failure counts (void -> int, five methods)", "status": "completed", "blockedBy": ["T2"] }, - { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "pending", "blockedBy": ["T3"] }, + { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "completed", "blockedBy": ["T3"] }, { "id": "T5", "subject": "Galaxy internal IMxWriteOps seam under GatewayGalaxyDataWriter (sealed SDK session untestable) (06/S-1)", "status": "pending", "blockedBy": [] }, { "id": "T6", "subject": "Galaxy fail-closed write on AdviseSupervisory failure -> BadCommunicationError, no raw write", "status": "pending", "blockedBy": ["T5"] }, { "id": "T7", "subject": "Galaxy counters galaxy.writes.advise_failed + galaxy.writes.unconfirmed (empty-statuses Good metered)", "status": "pending", "blockedBy": ["T6"] }, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index 668521fc..210d3705 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -58,6 +58,14 @@ public static class OtOpcUaTelemetry Meter.CreateCounter("otopcua.opcua.sink.write", unit: "{write}", description: "Writes that landed in IOpcUaAddressSpaceSink (kind=value|alarm|rebuild)."); + /// Address-space apply/materialise passes that swallowed at least one sink failure — + /// a failed rebuild (kind=rebuild) or per-node materialise failures (kind=nodes). A non-zero rate + /// means the running server holds a stale or partially-materialised address space despite a + /// reported-successful deploy (archreview 01/S-1). + public static readonly Counter OpcUaApplyFailed = + Meter.CreateCounter("otopcua.opcua.apply.failed", unit: "{apply}", + description: "Apply/materialise passes with swallowed sink failures (kind=rebuild|nodes)."); + public static readonly Counter ServiceLevelChange = Meter.CreateCounter("otopcua.redundancy.service_level_change", unit: "{change}", description: "OPC UA Server.ServiceLevel transitions emitted by the redundancy state."); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index a22ffcad..b6648b2d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -335,27 +335,43 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers var outcome = _applier.Apply(plan); _lastApplied = composition; - _applier.MaterialiseHierarchy(composition); + // Sum swallowed per-node materialise failures across every pass together with Apply's own + // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node + // Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter. + var failedNodes = outcome.FailedNodes; + failedNodes += _applier.MaterialiseHierarchy(composition); // T14 — scripted alarms get their own pass right after the hierarchy so the equipment // folders they parent under already exist. Materialises real Part 9 AlarmConditionState // nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled // alarms are skipped. - _applier.MaterialiseScriptedAlarms(composition); + failedNodes += _applier.MaterialiseScriptedAlarms(composition); // Equipment-namespace tags get their own pass: ensures each signal's Variable (and any // FolderPath sub-folder) exists under its already-materialised equipment folder so // clients can browse them. Live values are pushed by DriverHostActor.ForwardToMux after // each subscription cycle; variables show BadWaitingForInitialData only until the first // publish interval fires. - _applier.MaterialiseEquipmentTags(composition); + failedNodes += _applier.MaterialiseEquipmentTags(composition); // Equipment-namespace VirtualTags get their own pass right after the equipment tags: // ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its // equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live // values once the first dependency update arrives; until then variables show BadWaitingForInitialData. - _applier.MaterialiseEquipmentVirtualTags(composition); + failedNodes += _applier.MaterialiseEquipmentVirtualTags(composition); OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair("kind", "rebuild")); - _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", - msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); + if (outcome.RebuildFailed || failedNodes > 0) + { + OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, + new KeyValuePair("kind", outcome.RebuildFailed ? "rebuild" : "nodes")); + _log.Error( + "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial", + msg.Correlation, outcome.RebuildFailed, failedNodes, + outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes); + } + else + { + _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", + msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); + } } catch (Exception ex) { @@ -412,7 +428,16 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _log.Debug("OpcUaPublish: no applier wired — discarding MaterialiseDiscoveredNodes for {Equipment}", msg.EquipmentRootNodeId); return; } - _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables); + var failedNodes = _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables); + if (failedNodes > 0) + { + // archreview 01/S-1: a swallowed discovered-node injection failure surfaces at Error + the + // dedicated meter instead of vanishing into per-node Warnings. + OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair("kind", "nodes")); + _log.Error( + "OpcUaPublish: discovered-node injection DEGRADED for {Equipment} (failedNodes={FailedNodes}) — some discovered nodes may be missing", + msg.EquipmentRootNodeId, failedNodes); + } } private void HandleServiceLevelChanged(ServiceLevelChanged msg) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs new file mode 100644 index 00000000..2d401aaf --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs @@ -0,0 +1,170 @@ +using System.Diagnostics.Metrics; +using System.Text.Json; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +/// +/// archreview 01/S-1: when the applier reports a degraded apply (a rebuild that threw, or per-node +/// materialise failures), must surface it — increment the +/// otopcua.opcua.apply.failed meter and log at Error — instead of the optimistic Info line. The +/// happy path must stay Info-only (zero increments). +/// +public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase +{ + /// A rebuild whose sink throws increments otopcua.opcua.apply.failed (kind=rebuild). + [Fact] + public void Rebuild_when_sink_throws_increments_apply_failed_meter() + { + using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); + var db = NewInMemoryDbFactory(); + var sink = new ThrowOnRebuildSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + SeedEquipmentDeployment(db, "eq-1"); + + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + + AwaitAssert(() => + { + recorder.Total.ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("kind", "rebuild").ShouldBeGreaterThanOrEqualTo(1); + }, duration: TimeSpan.FromSeconds(2)); + } + + /// A clean rebuild does NOT increment the apply-failed meter (Info-only happy path). + [Fact] + public void Rebuild_happy_path_does_not_increment_apply_failed_meter() + { + using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); + var db = NewInMemoryDbFactory(); + var sink = new NoopSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + SeedEquipmentDeployment(db, "eq-1"); + + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + + // Give the actor time to process, then confirm the failed meter never fired. + Thread.Sleep(400); + recorder.Total.ShouldBe(0); + } + + private void AwaitAssert(Action assertion, TimeSpan duration) + { + var deadline = DateTime.UtcNow + duration; + Exception? last = null; + while (DateTime.UtcNow < deadline) + { + try { assertion(); return; } + catch (Exception ex) { last = ex; Thread.Sleep(25); } + } + if (last is not null) throw last; + } + + private static void SeedEquipmentDeployment(IDbContextFactory dbFactory, params string[] equipmentIds) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Equipment = equipmentIds.Select(id => new + { + EquipmentId = id, + MachineCode = id.ToUpperInvariant(), + UnsLineId = "line-1", + Name = id, + }).ToArray(), + DriverInstances = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + using var ctx = dbFactory.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = Guid.NewGuid(), + RevisionHash = new string('a', 64), + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + } + + /// A sink whose RebuildAddressSpace throws (drives the applier's SafeRebuild catch → RebuildFailed). + private sealed class ThrowOnRebuildSink : IOpcUaAddressSpaceSink + { + public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } + public void RebuildAddressSpace() => throw new InvalidOperationException("simulated rebuild fault"); + public void RaiseNodesAddedModelChange(string affectedNodeId) { } + } + + /// A no-op sink — a clean apply (no failures). + private sealed class NoopSink : IOpcUaAddressSpaceSink + { + public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } + public void RebuildAddressSpace() { } + public void RaiseNodesAddedModelChange(string affectedNodeId) { } + } + + /// Listens to a single instrument by name on the central meter and tallies value + tags. + private sealed class MeterRecorder : IDisposable + { + private readonly string _name; + private readonly MeterListener _listener; + private long _total; + private readonly List[]> _tagSets = new(); + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _name = instrumentName; + _listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name) + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + lock (_gate) + { + _total += value; + _tagSets.Add(tags.ToArray()); + } + }); + _listener.Start(); + } + + public long Total { get { lock (_gate) return _total; } } + + public int WithTag(string key, string value) + { + lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value))); + } + + public void Dispose() => _listener.Dispose(); + } +}