dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mesh): NodeCommandService on the telemetry listener

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-06 05:14:12 -04:00
parent 3f96115cab
commit 64774652b7
5 changed files with 310 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs
+110
View File
@@ -0,0 +1,110 @@
using Akka.Actor;
using Akka.Hosting;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Runtime;
// The generated service container is itself named NodeCommandService; alias its nested server base so
// this impl can keep the natural name without colliding with the generated type.
using GeneratedServiceBase =
ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1.NodeCommandService.NodeCommandServiceBase;
// Akka.Actor (imported for IActorRef.Tell) declares a Status of its own; only the gRPC one is used here.
using Status = Grpc.Core.Status;
namespace ZB.MOM.WW.OtOpcUa.Host.Grpc;
/// <summary>
/// Node-side command ingress (ClusterClient→gRPC migration Phase 1). Central dials a driver node
/// and calls <see cref="Execute"/>; the envelope is decoded back into the same CLR record the
/// ClusterClient receptionist used to deliver and handed to this node's
/// <c>NodeCommunicationActor</c>, which re-emits it on the node's <c>EventStream</c> exactly as
/// before. Nothing downstream of that actor can tell which transport carried the command.
/// </summary>
/// <remarks>
/// <para>
/// <b>One-way, verbatim from the transport it replaces.</b> <c>ExecuteAck</c> means "decoded and
/// handed to the comm actor" — not applied, and not acknowledged by any EventStream subscriber.
/// The <c>Tell</c> is deliberately not an <c>Ask</c>: no migrated command has a reply, and
/// waiting for one would invent a semantic central does not have.
/// </para>
/// <para>
/// <b>Decode runs BEFORE the registry lookup, and the two failures are graded differently.</b>
/// An undecodable envelope is <see cref="StatusCode.InvalidArgument"/> — permanent, and it will
/// never decode on a retry. A call arriving before <c>AddOtOpcUaRuntime</c> has spawned and
/// registered the comm actor is <see cref="StatusCode.Unavailable"/> — the house readiness
/// convention, transient, and correct to retry. Checking readiness first would report a
/// permanently malformed command as a readiness problem and invite an unbounded retry loop.
/// </para>
/// <para>
/// <b>Every log line carries the envelope's own <c>correlation_id</c>, read as a raw string.</b>
/// Three of the four commands also carry a domain correlation, but <c>AlarmCommand</c> has no
/// CLR member for one, so the envelope field is the only id every command shares — and it is
/// still readable when the inner fields are what failed to parse, which is precisely when an
/// operator needs it.
/// </para>
/// <para>
/// Inbound auth is <c>MeshCommandAuthInterceptor</c>, scoped to this service's method prefix and
/// registered at the <c>AddGrpc</c> level; it fail-closes with no <c>MeshTransport:GrpcPsk</c>
/// configured. Mapping this service is what makes that gate live.
/// </para>
/// </remarks>
public sealed class NodeCommandGrpcService : GeneratedServiceBase
{
private readonly ActorRegistry _registry;
private readonly ILogger<NodeCommandGrpcService> _log;
/// <summary>Initializes a new instance of the <see cref="NodeCommandGrpcService"/> class.</summary>
/// <param name="registry">Actor registry holding this node's <c>NodeCommunicationActor</c>.</param>
/// <param name="log">Logger.</param>
public NodeCommandGrpcService(ActorRegistry registry, ILogger<NodeCommandGrpcService> log)
{
_registry = registry;
_log = log;
}
/// <inheritdoc />
public override Task<ExecuteAck> Execute(CommandEnvelope request, ServerCallContext context)
{
ArgumentNullException.ThrowIfNull(request);
ArgumentNullException.ThrowIfNull(context);
object command;
try
{
command = MeshTransportDtoMapper.FromEnvelope(request);
}
catch (Exception ex) when (ex is InvalidOperationException or FormatException or ArgumentException)
{
// The envelope carries no credential material, so logging its correlation id is safe and is
// the only handle an operator has on a call whose inner fields did not parse.
_log.LogWarning(
ex,
"Rejected a mesh command (correlationId={CorrelationId}, oneof={CommandCase}): the "
+ "envelope could not be decoded. The command was NOT delivered.",
request.CorrelationId, request.CommandCase);
throw new RpcException(new Status(
StatusCode.InvalidArgument, "Command envelope could not be decoded."));
}
if (!_registry.TryGet<NodeCommunicationActorKey>(out var comm))
{
_log.LogWarning(
"Rejected a mesh command (correlationId={CorrelationId}, type={CommandType}): this node's "
+ "communication actor is not registered yet. The node is still starting; the command was "
+ "NOT delivered and central may retry.",
request.CorrelationId, command.GetType().Name);
throw new RpcException(new Status(
StatusCode.Unavailable, "Node is still starting: the command endpoint is not ready."));
}
comm.Tell(command);
_log.LogDebug(
"Mesh: accepted {CommandType} (correlationId={CorrelationId}) over gRPC; handed to the node "
+ "communication actor.",
command.GetType().Name, request.CorrelationId);
return Task.FromResult(new ExecuteAck());
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs
+9
View File
@@ -740,9 +740,18 @@ if (hasAdmin && configServeGrpcPort > 0)
// Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by
// TelemetryStreamAuthInterceptor (fail-closed on empty Telemetry:ApiKey).
//
// The node-side command endpoint (ClusterClient→gRPC migration Phase 1) rides the SAME driver-role
// h2c listener rather than claiming a fourth port: central already dials this node here, both
// directions are per-cluster-keyed, and a separate port would be one more thing to open on every
// site firewall for no isolation gain. Mapping it is what makes MeshCommandAuthInterceptor live —
// it is registered above and is a pass-through until a call matches this service's method prefix.
// It fail-closes with no MeshTransport:GrpcPsk, so an operator who maps the service without setting
// a key gets a closed endpoint, not an open one.
if (hasDriver && telemetryListenPort > 0)
{
app.MapGrpcService<TelemetryStreamGrpcService>();
app.MapGrpcService<NodeCommandGrpcService>();
}
app.MapOtOpcUaHealth();
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs
+2 -1
View File
@@ -2,5 +2,6 @@ using Xunit;
// The TelemetryStreamGrpcService concurrency counter is process-wide (static). Run this assembly's
// tests serially so the cap test observes a deterministic active-stream count with no cross-test
// interference. This project holds only telemetry tests, so serial execution costs nothing.
// interference. Every other suite here is small and fast (the mesh-command tests spin a TestKit
// ActorSystem per class), so serial execution costs nothing.
[assembly: CollectionBehavior(DisableTestParallelization = true)]
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs
+186
View File
@@ -0,0 +1,186 @@
using Akka.Hosting;
using Akka.TestKit.Xunit2;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Host.Grpc;
using ZB.MOM.WW.OtOpcUa.Runtime;
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Grpc;
/// <summary>
/// Verifies the node-side command ingress (ClusterClient→gRPC migration Phase 1): each of the four
/// central→node commands is decoded off its <see cref="CommandEnvelope"/> and handed to the
/// registered <c>NodeCommunicationActor</c> as the same CLR record the ClusterClient transport
/// delivered, an undecodable envelope is refused with <see cref="StatusCode.InvalidArgument"/>, and
/// a call landing in the boot window before the actor is registered is refused with
/// <see cref="StatusCode.Unavailable"/>.
/// </summary>
/// <remarks>
/// The actor is a <c>TestProbe</c> registered under <see cref="NodeCommunicationActorKey"/>: what
/// this service owes the rest of the system is the exact message the real actor already knows how to
/// re-emit, so asserting on what arrives at the registry key is the whole contract. The onward leg —
/// comm actor to EventStream to the real driver host — is pinned unchanged by
/// <c>EventStreamCommandDeliveryTests</c> in the Runtime test project.
/// </remarks>
public sealed class NodeCommandGrpcServiceTests : TestKit
{
private static readonly CorrelationId AlarmCorrelation = new(Guid.NewGuid());
private static ServerCallContext Ctx() => new FakeServerCallContext();
private static NodeCommandGrpcService NewService(ActorRegistry registry) =>
new(registry, NullLogger<NodeCommandGrpcService>.Instance);
[Fact]
public async Task Each_of_the_four_commands_reaches_the_comm_actor_as_its_domain_record()
{
var probe = CreateTestProbe();
var registry = new ActorRegistry();
registry.Register<NodeCommunicationActorKey>(probe.Ref);
var service = NewService(registry);
var dispatch = new DispatchDeployment(
new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid()));
var restart = new RestartDriver("cluster-a", "drv-1", "alice", Guid.NewGuid());
var reconnect = new ReconnectDriver("cluster-a", "drv-2", "bob", Guid.NewGuid());
var alarm = new AlarmCommand("alarm-7", "Acknowledge", "carol", "looked at it", null);
(await service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx())).ShouldNotBeNull();
probe.ExpectMsg<DispatchDeployment>().ShouldBe(dispatch);
await service.Execute(MeshTransportDtoMapper.ToEnvelope(restart), Ctx());
probe.ExpectMsg<RestartDriver>().ShouldBe(restart);
await service.Execute(MeshTransportDtoMapper.ToEnvelope(reconnect), Ctx());
probe.ExpectMsg<ReconnectDriver>().ShouldBe(reconnect);
await service.Execute(MeshTransportDtoMapper.ToEnvelope(alarm, AlarmCorrelation), Ctx());
probe.ExpectMsg<AlarmCommand>().ShouldBe(alarm);
}
[Fact]
public async Task A_call_in_the_boot_window_before_the_actor_is_registered_is_unavailable()
{
// Nothing registered: the ActorSystem exists but AddOtOpcUaRuntime has not spawned the comm
// actor yet. Unavailable (not Internal) because the condition is transient and retryable —
// central's dialer may try the same node again a moment later and succeed.
var service = NewService(new ActorRegistry());
var dispatch = new DispatchDeployment(
new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid()));
var ex = await Should.ThrowAsync<RpcException>(
() => service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx()));
ex.StatusCode.ShouldBe(StatusCode.Unavailable);
}
[Fact]
public async Task An_unset_oneof_is_invalid_argument()
{
var probe = CreateTestProbe();
var registry = new ActorRegistry();
registry.Register<NodeCommunicationActorKey>(probe.Ref);
var service = NewService(registry);
var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(
new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
[Theory]
[MemberData(nameof(UndecodableEnvelopes))]
public async Task An_envelope_whose_fields_do_not_parse_is_invalid_argument(CommandEnvelope envelope)
{
var probe = CreateTestProbe();
var registry = new ActorRegistry();
registry.Register<NodeCommunicationActorKey>(probe.Ref);
var service = NewService(registry);
var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(envelope, Ctx()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
public static TheoryData<CommandEnvelope> UndecodableEnvelopes() =>
[
// Deployment id is not a hyphenless GUID -> FormatException out of the mapper.
new CommandEnvelope
{
CorrelationId = Guid.NewGuid().ToString("N"),
DispatchDeployment = new DispatchDeploymentCommand
{
DeploymentId = "not-a-guid",
RevisionHash = "rev-1",
CorrelationId = Guid.NewGuid().ToString("N"),
},
},
// Revision hash is blank -> ArgumentException out of RevisionHash.Parse.
new CommandEnvelope
{
CorrelationId = Guid.NewGuid().ToString("N"),
DispatchDeployment = new DispatchDeploymentCommand
{
DeploymentId = Guid.NewGuid().ToString("N"),
RevisionHash = "",
CorrelationId = Guid.NewGuid().ToString("N"),
},
},
// Restart command with an unparseable correlation id -> FormatException.
new CommandEnvelope
{
CorrelationId = Guid.NewGuid().ToString("N"),
RestartDriver = new RestartDriverCommand
{
ClusterId = "cluster-a",
DriverInstanceId = "drv-1",
ActorByUserName = "alice",
CorrelationId = "nope",
},
},
];
[Fact]
public async Task An_undecodable_envelope_in_the_boot_window_is_invalid_argument_not_unavailable()
{
// Ordering pin: decode runs BEFORE the registry lookup. Unavailable invites the caller to
// retry, and an envelope this build cannot decode will never decode — reporting it as a
// readiness problem would turn a permanent fault into an unbounded retry loop.
var service = NewService(new ActorRegistry());
var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(
new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
}
/// <summary>Minimal <see cref="ServerCallContext"/>; this RPC reads nothing off the context but its method name.</summary>
private sealed class FakeServerCallContext : ServerCallContext
{
protected override string MethodCore => "/mesh_transport.v1.NodeCommandService/Execute";
protected override string HostCore => "localhost";
protected override string PeerCore => "ipv4:127.0.0.1:0";
protected override DateTime DeadlineCore => DateTime.MaxValue;
protected override Metadata RequestHeadersCore => [];
protected override CancellationToken CancellationTokenCore => CancellationToken.None;
protected override Metadata ResponseTrailersCore { get; } = [];
protected override Status StatusCore { get; set; }
protected override WriteOptions? WriteOptionsCore { get; set; }
protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj
+3
View File
@@ -10,6 +10,9 @@
<ItemGroup>
<PackageReference Include="xunit"/>
<PackageReference Include="Shouldly"/>
<!-- NodeCommandGrpcServiceTests registers a TestProbe under NodeCommunicationActorKey, so the
service is asserted against the same ActorRegistry seam the Host wires in production. -->
<PackageReference Include="Akka.TestKit.Xunit2"/>
<!-- ConfigDbApiKeyStoreTests drives the store against a real EF query, the same in-memory
provider convention the ControlPlane/Runtime test projects use. No central SQL Server. -->
<PackageReference Include="Microsoft.EntityFrameworkCore.InMemory"/>