Joseph Doherty
426 lines
18 KiB
C#
426 lines
18 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Diagnostics.Metrics;
|
|
using System.Text.Json;
|
|
using Akka.Actor;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.TestKit;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
|
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.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
|
|
|
/// <summary>
|
|
/// archreview 03/S4 — the boot-window primary-gate guard. On a MULTI-driver cluster (member count > 1)
|
|
/// while the role is UNKNOWN, the Primary data-plane gates (inbound write, native ack, native alerts emit)
|
|
/// default-DENY; on a SINGLE-driver cluster (count <= 1) they default-ALLOW (boot-window / single-node
|
|
/// posture). Driver member count is injected via the <c>driverMemberCountProvider</c> Props seam.
|
|
/// </summary>
|
|
public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase
|
|
{
|
|
private static readonly NodeId TestNode = NodeId.Parse("driver-pgate-test");
|
|
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
|
|
private static readonly DateTime Ts = new(2026, 7, 13, 10, 0, 0, DateTimeKind.Utc);
|
|
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
|
|
|
|
// ---------------- write gate ----------------
|
|
|
|
/// <summary>Role unknown + a real driver peer (count 2) ⇒ RouteNodeWrite is DENIED with the
|
|
/// boot-window reason and the driver sees no write.</summary>
|
|
[Fact]
|
|
public void Unknown_role_multi_driver_denies_write_with_role_unknown_reason()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var recorder = new RecordingDriverFactory("Modbus");
|
|
var dep = SeedWriteTagDeployment(db);
|
|
var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2);
|
|
|
|
var asker = CreateTestProbe();
|
|
actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref);
|
|
|
|
var result = asker.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
|
result.Success.ShouldBeFalse();
|
|
result.Reason.ShouldBe("not primary (role unknown)");
|
|
recorder.Writes.ShouldBeEmpty();
|
|
}
|
|
|
|
/// <summary>Role unknown + no driver peer (count 1) ⇒ the write is SERVICED (single-node / boot-window
|
|
/// posture preserved).</summary>
|
|
[Fact]
|
|
public void Unknown_role_single_driver_services_write()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var recorder = new RecordingDriverFactory("Modbus");
|
|
var dep = SeedWriteTagDeployment(db);
|
|
var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 1);
|
|
|
|
// The gate must ALLOW (count 1, role unknown). The driver child's async connect can lag the ApplyAck,
|
|
// so a first write may transiently report "driver not connected" — retry until connected (a real
|
|
// client retries too). The point under test is that the gate never denies (never "not primary").
|
|
AwaitAssert(() =>
|
|
{
|
|
var asker = CreateTestProbe();
|
|
actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref);
|
|
var res = asker.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
|
res.Reason.ShouldNotBe("not primary");
|
|
res.Reason.ShouldNotBe("not primary (role unknown)");
|
|
res.Success.ShouldBeTrue(res.Reason);
|
|
}, duration: TimeSpan.FromSeconds(10));
|
|
recorder.Writes.ShouldNotBeEmpty();
|
|
}
|
|
|
|
/// <summary>A Primary snapshot services even at count 2; a Secondary snapshot denies with the steady-state
|
|
/// reason (distinct from the boot-window reason).</summary>
|
|
[Fact]
|
|
public void Known_role_wins_over_member_count()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var recorder = new RecordingDriverFactory("Modbus");
|
|
var dep = SeedWriteTagDeployment(db);
|
|
var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2);
|
|
|
|
// Primary snapshot ⇒ serviced even with a driver peer. Retry through the transient driver-connect
|
|
// window; the invariant under test is that the gate never denies a Primary.
|
|
TellRole(actor, RedundancyRole.Primary);
|
|
AwaitAssert(() =>
|
|
{
|
|
var asker1 = CreateTestProbe();
|
|
actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 1.0), asker1.Ref);
|
|
var res = asker1.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
|
res.Reason.ShouldNotBe("not primary");
|
|
res.Success.ShouldBeTrue(res.Reason);
|
|
}, duration: TimeSpan.FromSeconds(10));
|
|
|
|
// Secondary snapshot ⇒ denied with the steady-state reason.
|
|
TellRole(actor, RedundancyRole.Secondary);
|
|
var asker2 = CreateTestProbe();
|
|
actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 2.0), asker2.Ref);
|
|
var denied = asker2.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
|
denied.Success.ShouldBeFalse();
|
|
denied.Reason.ShouldBe("not primary");
|
|
}
|
|
|
|
/// <summary>The denial meter carries the site + reason tags.</summary>
|
|
[Fact]
|
|
public void Denied_write_increments_primary_gate_denied_meter_with_tags()
|
|
{
|
|
using var recorder = new MeterRecorder("otopcua.redundancy.primary_gate_denied");
|
|
var db = NewInMemoryDbFactory();
|
|
var factory = new RecordingDriverFactory("Modbus");
|
|
var dep = SeedWriteTagDeployment(db);
|
|
var actor = SpawnWriteHost(db, dep, factory, driverMemberCount: 2);
|
|
|
|
var asker = CreateTestProbe();
|
|
actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 9.0), asker.Ref);
|
|
asker.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout).Success.ShouldBeFalse();
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
recorder.Total.ShouldBeGreaterThanOrEqualTo(1);
|
|
recorder.WithTag("site", "write").ShouldBeGreaterThanOrEqualTo(1);
|
|
recorder.WithTag("reason", "role-unknown").ShouldBeGreaterThanOrEqualTo(1);
|
|
}, duration: Timeout);
|
|
}
|
|
|
|
// ---------------- ack gate ----------------
|
|
|
|
/// <summary>Role unknown + count 2 ⇒ a native-alarm ack is dropped at Warning and the denial meter
|
|
/// increments with site=ack.</summary>
|
|
[Fact]
|
|
public void Unknown_role_multi_driver_drops_native_ack_with_warning_and_meter()
|
|
{
|
|
using var meter = new MeterRecorder("otopcua.redundancy.primary_gate_denied");
|
|
var db = NewInMemoryDbFactory();
|
|
var dep = SeedAlarmTagDeployment(db);
|
|
var (actor, _) = SpawnAlarmHost(db, dep, driverMemberCount: 2);
|
|
|
|
EventFilter.Warning(contains: "role unknown").ExpectOne(() =>
|
|
actor.Tell(new DriverHostActor.RouteNativeAlarmAck("eq-1/temp_hi", Comment: null, OperatorUser: "op")));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
meter.WithTag("site", "ack").ShouldBeGreaterThanOrEqualTo(1);
|
|
}, duration: Timeout);
|
|
}
|
|
|
|
// ---------------- alarm-emit gate ----------------
|
|
|
|
/// <summary>Role unknown + count 2 ⇒ ForwardNativeAlarm still writes the (ungated) OPC UA condition update
|
|
/// but publishes NO cluster-wide alerts transition.</summary>
|
|
[Fact]
|
|
public void Unknown_role_multi_driver_suppresses_alerts_emit_but_updates_condition()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var dep = SeedAlarmTagDeployment(db);
|
|
var alerts = CreateTestProbe();
|
|
SubscribeToAlerts(alerts);
|
|
var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 2);
|
|
|
|
actor.Tell(RaiseAlarm());
|
|
|
|
// The ungated OPC UA condition write still arrives.
|
|
publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout).AlarmNodeId.ShouldBe("eq-1/temp_hi");
|
|
// The cluster-wide alerts publish is suppressed (a real Primary peer publishes the single copy).
|
|
alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Role unknown + count 1 ⇒ ForwardNativeAlarm publishes the alerts transition (single-node
|
|
/// posture preserved).</summary>
|
|
[Fact]
|
|
public void Unknown_role_single_driver_publishes_alerts_emit()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var dep = SeedAlarmTagDeployment(db);
|
|
var alerts = CreateTestProbe();
|
|
SubscribeToAlerts(alerts);
|
|
var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 1);
|
|
|
|
actor.Tell(RaiseAlarm());
|
|
|
|
publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout);
|
|
alerts.ExpectMsg<AlarmTransitionEvent>(Timeout).AlarmId.ShouldBe("eq-1/temp_hi");
|
|
}
|
|
|
|
// ---------------- helpers ----------------
|
|
|
|
private static DriverInstanceActor.AttributeAlarmPublished RaiseAlarm() =>
|
|
new("drv-1", new AlarmEventArgs(
|
|
new StubAlarmHandle(),
|
|
SourceNodeId: "Temp",
|
|
ConditionId: "Temp.HiHi",
|
|
AlarmType: "OffNormalAlarm",
|
|
Message: "temperature high",
|
|
Severity: AlarmSeverity.High,
|
|
SourceTimestampUtc: Ts,
|
|
Kind: AlarmTransitionKind.Raise));
|
|
|
|
private static void TellRole(IActorRef host, RedundancyRole role) =>
|
|
host.Tell(new RedundancyStateChanged(
|
|
new[]
|
|
{
|
|
new NodeRedundancyState(TestNode, role,
|
|
IsClusterLeader: role == RedundancyRole.Primary,
|
|
IsRoleLeaderForDriver: role == RedundancyRole.Primary,
|
|
AsOfUtc: DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
private IActorRef SpawnWriteHost(
|
|
IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId dep, IDriverFactory factory, int driverMemberCount)
|
|
{
|
|
var coordinator = CreateTestProbe();
|
|
var actor = Sys.ActorOf(DriverHostActor.Props(
|
|
db, TestNode, coordinator.Ref,
|
|
driverFactory: factory,
|
|
localRoles: new HashSet<string> { "driver" },
|
|
driverMemberCountProvider: () => driverMemberCount));
|
|
|
|
actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId()));
|
|
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
|
return actor;
|
|
}
|
|
|
|
private (IActorRef Actor, TestProbe Publish) SpawnAlarmHost(
|
|
IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId dep, int driverMemberCount)
|
|
{
|
|
var coordinator = CreateTestProbe();
|
|
var publish = CreateTestProbe();
|
|
var mux = CreateTestProbe();
|
|
var vtHost = CreateTestProbe();
|
|
|
|
var actor = Sys.ActorOf(DriverHostActor.Props(
|
|
db, TestNode, coordinator.Ref,
|
|
localRoles: new HashSet<string> { "driver" },
|
|
dependencyMux: mux.Ref,
|
|
opcUaPublishActor: publish.Ref,
|
|
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
|
|
virtualTagHostOverride: vtHost.Ref,
|
|
driverMemberCountProvider: () => driverMemberCount));
|
|
|
|
actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId()));
|
|
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
|
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(Timeout);
|
|
return (actor, publish);
|
|
}
|
|
|
|
private void SubscribeToAlerts(TestProbe probe)
|
|
{
|
|
DistributedPubSub.Get(Sys).Mediator.Tell(
|
|
new Subscribe(ScriptedAlarmHostActor.AlertsTopic, probe.Ref), probe.Ref);
|
|
probe.ExpectMsg<SubscribeAck>(Timeout);
|
|
}
|
|
|
|
private static DeploymentId SeedWriteTagDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db)
|
|
{
|
|
var artifact = JsonSerializer.SerializeToUtf8Bytes(new
|
|
{
|
|
Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } },
|
|
DriverInstances = new[]
|
|
{
|
|
new
|
|
{
|
|
DriverInstanceRowId = Guid.NewGuid(),
|
|
DriverInstanceId = "drv-1",
|
|
Name = "drv-1",
|
|
DriverType = "Modbus",
|
|
Enabled = true,
|
|
DriverConfig = "{}",
|
|
NamespaceId = "ns-eq",
|
|
},
|
|
},
|
|
Tags = new[]
|
|
{
|
|
new
|
|
{
|
|
TagId = "tag-0",
|
|
EquipmentId = "eq-1",
|
|
DriverInstanceId = "drv-1",
|
|
Name = "speed",
|
|
FolderPath = (string?)null,
|
|
DataType = "Double",
|
|
TagConfig = JsonSerializer.Serialize(new { FullName = "40001" }),
|
|
},
|
|
},
|
|
});
|
|
return SealArtifact(db, artifact);
|
|
}
|
|
|
|
private static DeploymentId SeedAlarmTagDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db)
|
|
{
|
|
var artifact = JsonSerializer.SerializeToUtf8Bytes(new
|
|
{
|
|
Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } },
|
|
DriverInstances = new[] { new { DriverInstanceId = "drv-1", NamespaceId = "ns-eq" } },
|
|
Tags = new[]
|
|
{
|
|
new
|
|
{
|
|
TagId = "tag-0",
|
|
EquipmentId = "eq-1",
|
|
DriverInstanceId = "drv-1",
|
|
Name = "temp_hi",
|
|
FolderPath = (string?)null,
|
|
DataType = "Boolean",
|
|
TagConfig = JsonSerializer.Serialize(new
|
|
{
|
|
FullName = "Temp.HiHi",
|
|
alarm = new { alarmType = "OffNormalAlarm", severity = 700 },
|
|
}),
|
|
},
|
|
},
|
|
});
|
|
return SealArtifact(db, artifact);
|
|
}
|
|
|
|
private static DeploymentId SealArtifact(IDbContextFactory<OtOpcUaConfigDbContext> db, byte[] artifact)
|
|
{
|
|
var id = DeploymentId.NewId();
|
|
using var ctx = db.CreateDbContext();
|
|
ctx.Deployments.Add(new Deployment
|
|
{
|
|
DeploymentId = id.Value,
|
|
RevisionHash = RevA.Value,
|
|
Status = DeploymentStatus.Sealed,
|
|
CreatedBy = "test",
|
|
SealedAtUtc = DateTime.UtcNow,
|
|
ArtifactBlob = artifact,
|
|
});
|
|
ctx.SaveChanges();
|
|
return id;
|
|
}
|
|
|
|
private sealed class StubAlarmHandle : IAlarmSubscriptionHandle
|
|
{
|
|
public string DiagnosticId => "stub-alarm-sub";
|
|
}
|
|
|
|
private sealed class RecordingDriverFactory : IDriverFactory
|
|
{
|
|
private readonly string _supportedType;
|
|
private readonly RecordingDriver _driver = new();
|
|
public RecordingDriverFactory(string supportedType) { _supportedType = supportedType; }
|
|
public IReadOnlyList<WriteRequest> Writes => _driver.Writes;
|
|
|
|
public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson)
|
|
{
|
|
if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null;
|
|
_driver.Bind(driverInstanceId, driverType);
|
|
return _driver;
|
|
}
|
|
|
|
public IReadOnlyCollection<string> SupportedTypes => new[] { _supportedType };
|
|
}
|
|
|
|
private sealed class RecordingDriver : IDriver, IWritable
|
|
{
|
|
private readonly ConcurrentQueue<WriteRequest> _writes = new();
|
|
public string DriverInstanceId { get; private set; } = string.Empty;
|
|
public string DriverType { get; private set; } = string.Empty;
|
|
public IReadOnlyList<WriteRequest> Writes => _writes.ToArray();
|
|
public void Bind(string id, string type) { DriverInstanceId = id; DriverType = type; }
|
|
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, DateTime.UtcNow, LastError: null);
|
|
public long GetMemoryFootprint() => 0;
|
|
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
|
public Task<IReadOnlyList<WriteResult>> WriteAsync(
|
|
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
|
{
|
|
foreach (var w in writes) _writes.Enqueue(w);
|
|
return Task.FromResult<IReadOnlyList<WriteResult>>(writes.Select(_ => new WriteResult(0u)).ToArray());
|
|
}
|
|
}
|
|
|
|
private sealed class MeterRecorder : IDisposable
|
|
{
|
|
private readonly string _name;
|
|
private readonly MeterListener _listener;
|
|
private long _total;
|
|
private readonly List<KeyValuePair<string, object?>[]> _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<long>((_, 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();
|
|
}
|
|
}
|