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

feat(mesh): Grpc mode value + validation

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-06 03:00:59 -04:00
parent c0ccfa27fb
commit aa21f61fa0
7 changed files with 521 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs
+109 -8
View File
@@ -6,11 +6,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster;
/// </summary>
/// <remarks>
/// <para>
/// <see cref="Mode"/> is a deliberate <b>dark switch</b>, not a permanent knob: both
/// transports live in the tree for one phase so the docker-dev rig can run the same
/// deployment over each and compare, and so a bad Phase 2 rolls back with a config change
/// rather than a revert. Phase 6 deletes the DistributedPubSub branch along with the single
/// mesh that made it viable.
/// <see cref="Mode"/> is a deliberate <b>dark switch</b>: the transports live in the tree
/// side by side so the docker-dev rig can run the same deployment over each and compare, and
/// so a bad rollout rolls back with a config change rather than a revert.
/// </para>
/// <para>
/// <b>The DistributedPubSub branch is retained, not scheduled for deletion.</b> An earlier
/// comment here promised Phase 6 would delete it along with the single mesh; that was
/// deliberately reversed — see <c>docs/Configuration.md</c> § "The compiled transport-mode
/// defaults were deliberately NOT flipped". DPS remains the correct and default transport for
/// a single-mesh topology, which every integration test and every profile that leaves this
/// key unset still runs. What is being replaced is <see cref="ModeClusterClient"/>, by
/// <see cref="ModeGrpc"/>.
/// </para>
/// <para>
/// <b>Discovery is deliberately asymmetric</b>, matching the sister project. Central
@@ -32,9 +39,19 @@ public sealed class MeshTransportOptions
public const string ModeClusterClient = "ClusterClient";
/// <summary>
/// <see cref="ModeDps"/> (default) or <see cref="ModeClusterClient"/>. Any other value fails
/// the host at startup rather than silently falling back — a transport that is neither of
/// these delivers nothing, and the symptom would be commands that vanish.
/// Commands cross the mesh boundary as unary gRPC calls on the <c>mesh_transport.v1</c>
/// contract — central dials each node's <c>NodeCommandService</c>, a node posts its
/// deployment ack to central's <c>CentralCommandService</c>. The replacement for
/// <see cref="ModeClusterClient"/>; unlike it, the command plane is authenticated
/// (<see cref="GrpcPsk"/>).
/// </summary>
public const string ModeGrpc = "Grpc";
/// <summary>
/// <see cref="ModeDps"/> (default), <see cref="ModeClusterClient"/> or
/// <see cref="ModeGrpc"/>. Any other value fails the host at startup rather than silently
/// falling back — a transport that is none of these delivers nothing, and the symptom would
/// be commands that vanish.
/// </summary>
public string Mode { get; set; } = ModeDps;
@@ -58,4 +75,88 @@ public sealed class MeshTransportOptions
/// process — migration, direct SQL, a second admin node — is picked up without a restart.
/// </summary>
public int ContactRefreshSeconds { get; set; } = 60;
/// <summary>
/// <c>MeshTransport:CentralGrpcEndpoints</c> — node-side only, and only under
/// <see cref="ModeGrpc"/>: the gRPC base addresses of the central pair's
/// <c>CentralCommandService</c>, e.g. <c>http://central-1:4055</c>. Full URLs
/// (<c>scheme://host:port</c>), not bare <c>host:port</c> — <c>http://</c> is h2c.
/// </summary>
/// <remarks>
/// Tried <b>in order, stopping at the first success</b>, which is what preserves the
/// receptionist's one-ack-per-node-per-dispatch semantic: either central node routes the ack
/// to the same <c>ConfigPublishCoordinator</c> singleton proxy, so which one answers does not
/// matter, but two answering would double-count. The gRPC counterpart of
/// <see cref="CentralContactPoints"/>, and it replaces it — a node running
/// <see cref="ModeGrpc"/> never reads the Akka contact points.
/// </remarks>
public string[] CentralGrpcEndpoints { get; set; } = [];
/// <summary>
/// <c>MeshTransport:GrpcPsk</c> — this node's pre-shared key for the command plane, one per
/// application <c>Cluster</c>. Under <see cref="ModeGrpc"/> it is both the bearer this node
/// presents when posting an ack to central and the bearer it requires on inbound
/// <c>NodeCommandService</c> calls.
/// </summary>
/// <remarks>
/// <para>
/// Production form is a secret reference — <c>${secret:OT-MESH-PSK-&lt;clusterId&gt;}</c>,
/// e.g. <c>${secret:OT-MESH-PSK-SITE-A}</c> — resolved pre-host by the existing
/// <c>SecretReferenceExpander</c>, so it works on a driver-only node with no ConfigDb.
/// Supply a literal only on a rig, via the environment
/// (<c>MeshTransport__GrpcPsk</c>); never commit one.
/// </para>
/// <para>
/// <b>Per-Cluster, not fleet-wide, on purpose.</b> The three existing shared-bearer gRPC
/// channels are read/telemetry planes; this one carries <c>RestartDriver</c>,
/// <c>AlarmCommand</c> and deployment dispatch, so a single fleet key would let one
/// compromised site node command every other site and forge acks fleet-wide.
/// </para>
/// <para>
/// Nullable because <b>unset must be distinguishable from blank at the boundary that
/// fails closed</b>: the serve-side interceptor rejects every call when no key is
/// configured rather than defaulting to an empty comparison that any empty bearer would
/// satisfy.
/// </para>
/// </remarks>
public string? GrpcPsk { get; set; }
/// <summary>
/// <c>MeshTransport:DispatchTimeoutSeconds</c> — central-side deadline, in seconds, for one
/// <c>NodeCommandService.Execute</c> call to one node. Must be greater than 0 under
/// <see cref="ModeGrpc"/>.
/// </summary>
/// <remarks>
/// Bounded per node because a fan-out dispatch dials every enabled node in the cluster: an
/// unbounded call to one unreachable node would stall the commands queued behind it. Expiry
/// is logged and dropped — the transport is fire-and-forget and layers no retry, so the
/// deploy path's 2-minute apply deadline stays the only recovery mechanism.
/// </remarks>
public int DispatchTimeoutSeconds { get; set; } = 10;
/// <summary>
/// <c>MeshTransport:AckTimeoutSeconds</c> — node-side deadline, in seconds, for one
/// <c>CentralCommandService.ReportApply</c> call to one central endpoint. Must be greater
/// than 0 under <see cref="ModeGrpc"/>.
/// </summary>
/// <remarks>
/// Applied per endpoint, not per ack: the node walks <see cref="CentralGrpcEndpoints"/> in
/// order, so a dead first endpoint costs this deadline before the second is tried.
/// </remarks>
public int AckTimeoutSeconds { get; set; } = 10;
/// <summary>
/// <c>MeshTransport:ClusterPsks:{clusterId}</c> — central-side override map of per-cluster
/// PSKs, consulted <i>before</i> the Secrets store when central builds a channel to (or
/// verifies an ack from) a given <c>Cluster</c>. Keyed case-insensitively.
/// </summary>
/// <remarks>
/// <b>Production leaves this empty</b> and the runtime provider resolves
/// <c>OT-MESH-PSK-{clusterId}</c> from the Secrets store instead; the map exists so the
/// docker-dev rig and integration tests can supply literals through the environment
/// (<c>MeshTransport__ClusterPsks__SITE-A</c>) without standing up a KEK. Central needs a map
/// rather than a single key because its cluster set is DB-dynamic — one key per site pair it
/// talks to — whereas a node holds exactly its own (<see cref="GrpcPsk"/>).
/// </remarks>
public Dictionary<string, string> ClusterPsks { get; set; } = new(StringComparer.OrdinalIgnoreCase);
}
src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs
+93 -3
View File
@@ -1,3 +1,4 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.OtOpcUa.Cluster;
@@ -15,6 +16,22 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster;
/// </remarks>
public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTransportOptions>
{
private readonly IConfiguration _configuration;
/// <summary>
/// DI-constructed by <c>AddValidatedOptions</c> (a plain <c>AddSingleton</c>), so a
/// constructor dependency is safe here. <see cref="IConfiguration"/> — not
/// <c>IClusterRoleInfo</c> — is the source of this node's roles: <c>IClusterRoleInfo</c>'s
/// implementation needs the <c>ActorSystem</c>, which does not exist yet at
/// <c>ValidateOnStart</c> time. Same shape as <see cref="ConfigSourceOptionsValidator"/> and
/// <see cref="TelemetryOptionsValidator"/>.
/// </summary>
/// <param name="configuration">The application configuration, read for <c>Cluster:Roles</c>.</param>
public MeshTransportOptionsValidator(IConfiguration configuration)
{
_configuration = configuration;
}
/// <inheritdoc />
public ValidateOptionsResult Validate(string? name, MeshTransportOptions options)
{
@@ -25,13 +42,14 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTranspo
var isDps = string.Equals(options.Mode, MeshTransportOptions.ModeDps, StringComparison.OrdinalIgnoreCase);
var isClusterClient = string.Equals(
options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
var isGrpc = string.Equals(options.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);
if (!isDps && !isClusterClient)
if (!isDps && !isClusterClient && !isGrpc)
{
errors.Add(
$"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.Mode)} is "
+ $"'{options.Mode}'. Expected '{MeshTransportOptions.ModeDps}' or "
+ $"'{MeshTransportOptions.ModeClusterClient}'.");
+ $"'{options.Mode}'. Expected '{MeshTransportOptions.ModeDps}', "
+ $"'{MeshTransportOptions.ModeClusterClient}' or '{MeshTransportOptions.ModeGrpc}'.");
}
if (options.ContactRefreshSeconds <= 0)
@@ -76,6 +94,78 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTranspo
}
}
// The Grpc surface below is node-side: the PSK a driver presents (and requires), the central
// endpoints it posts its ack to, and the two call deadlines. Scoped to the driver role for the
// same reason the ClusterClient block above is scoped to its mode — an admin-only central node
// dials nodes rather than being dialled and resolves per-cluster keys at runtime through
// MeshTransportOptions.ClusterPsks / the Secrets store, so requiring a single GrpcPsk and an
// endpoint list of it would demand configuration it never reads. Roles come from
// configuration, not IClusterRoleInfo (see the ctor remark), and through
// RoleParser.ResolveEffectiveRoles so the OTOPCUA_ROLES fallback applies (Gitea #532) — a node
// configured only through the env var is still the driver that has to dial central, and
// exempting it here would leave its acks rejected unauthenticated at runtime.
var roles = RoleParser.ResolveEffectiveRoles(
_configuration.GetSection("Cluster:Roles").Get<string[]>());
var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0;
if (isGrpc && isDriver)
{
if (string.IsNullOrWhiteSpace(options.GrpcPsk))
{
errors.Add(
$"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.GrpcPsk)} is "
+ $"empty. Under {MeshTransportOptions.ModeGrpc} this per-cluster pre-shared key is "
+ "the whole authentication boundary of the command plane in both directions — "
+ "without it this node's inbound command service fails closed and every ack it "
+ "posts to central is rejected. Supply it via the environment "
+ $"({MeshTransportOptions.SectionName}__{nameof(MeshTransportOptions.GrpcPsk)}), in "
+ "production as ${secret:OT-MESH-PSK-<clusterId>}.");
}
if (options.CentralGrpcEndpoints.Length == 0)
{
errors.Add(
$"{MeshTransportOptions.SectionName}:"
+ $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)} is empty. Under "
+ $"{MeshTransportOptions.ModeGrpc} a driver node has nowhere to post its "
+ "deployment ack, and every deployment would time out naming a node that is "
+ "running perfectly. Supply them via the environment "
+ $"({MeshTransportOptions.SectionName}__"
+ $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)}__0).");
}
foreach (var endpoint in options.CentralGrpcEndpoints)
{
if (!endpoint.StartsWith("http://", StringComparison.OrdinalIgnoreCase)
&& !endpoint.StartsWith("https://", StringComparison.OrdinalIgnoreCase))
{
errors.Add(
$"Central gRPC endpoint '{endpoint}' must start with 'http://' (h2c) or "
+ "'https://' — it is a gRPC base address (scheme://host:port), not a bare "
+ "host:port and not an Akka remoting address.");
}
}
// Both deadlines are per-call and there is no retry behind them, so a non-positive value
// is not "wait forever" — it is a call that expires before it is made, i.e. a command
// plane that silently delivers nothing.
if (options.DispatchTimeoutSeconds <= 0)
{
errors.Add(
$"{MeshTransportOptions.SectionName}:"
+ $"{nameof(MeshTransportOptions.DispatchTimeoutSeconds)} must be greater than 0 "
+ $"(was {options.DispatchTimeoutSeconds}).");
}
if (options.AckTimeoutSeconds <= 0)
{
errors.Add(
$"{MeshTransportOptions.SectionName}:"
+ $"{nameof(MeshTransportOptions.AckTimeoutSeconds)} must be greater than 0 "
+ $"(was {options.AckTimeoutSeconds}).");
}
}
return errors.Count == 0
? ValidateOptionsResult.Success
: ValidateOptionsResult.Fail(string.Join(" ", errors));
src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs
+10 -2
View File
@@ -98,14 +98,22 @@ public sealed class SplitTopologyTransportValidator : IValidateOptions<MeshTrans
var telemetryMode = _configuration["Telemetry:Mode"] ?? TelemetryOptions.ModeDps;
var telemetryDialMode = _configuration["TelemetryDial:Mode"] ?? TelemetryDialOptions.ModeDps;
if (!string.Equals(meshMode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase))
// Two accepted values, not one, for the duration of the ClusterClient→gRPC migration: both
// cross a mesh boundary, which is the only property this rule cares about. Without the
// relaxation the very first rollout step — flipping a split-topology node to Grpc — would be a
// host-start failure. It is NOT widened to "anything but Dps": a typo'd mode delivers nothing
// just as surely, so the check stays an explicit allow-list and tightens to Grpc-only once
// ClusterClient is deleted.
if (!string.Equals(meshMode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase)
&& !string.Equals(meshMode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase))
{
errors.Add(
$"Cluster:Roles declares a cluster-{clusterId} role (split topology) but "
+ $"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.Mode)} is "
+ $"'{meshMode}'; a split mesh cannot use DistributedPubSub — set "
+ $"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.Mode)}="
+ $"{MeshTransportOptions.ModeClusterClient}.");
+ $"{MeshTransportOptions.ModeGrpc} (or {MeshTransportOptions.ModeClusterClient}, the "
+ "transport it replaces).");
}
if (isDriver
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs
+6 -2
View File
@@ -88,8 +88,12 @@ public sealed class AdminOperationsActor : ReceiveActor
/// </summary>
/// <remarks>
/// The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh Phase 2)
/// on exactly the pre-Phase-2 behaviour. It is not a permanent seam: Phase 6 deletes it along
/// with the DistributedPubSub branch and the single mesh.
/// on exactly the pre-Phase-2 behaviour. The DistributedPubSub branch behind it is
/// <b>retained</b>, not scheduled for deletion — an earlier note here promised Phase 6 would
/// remove it, which was deliberately reversed (<c>docs/Configuration.md</c> § "The compiled
/// transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and
/// default transport for a single-mesh topology. What is being replaced is the ClusterClient
/// branch, by gRPC.
/// </remarks>
/// <param name="topic">The legacy DistributedPubSub topic.</param>
/// <param name="message">The command.</param>
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs
+5 -2
View File
@@ -68,8 +68,11 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// </summary>
/// <remarks>
/// Null in tests and on any node wired before Phase 2, where the fallback reproduces the
/// pre-Phase-2 behaviour exactly. The fallback is not a permanent seam — Phase 6 deletes it
/// along with the DistributedPubSub branch and the single mesh.
/// pre-Phase-2 behaviour exactly. The DistributedPubSub branch behind it is <b>retained</b>,
/// not scheduled for deletion — an earlier note here promised Phase 6 would remove it, which
/// was deliberately reversed (<c>docs/Configuration.md</c> § "The compiled transport-mode
/// defaults were deliberately NOT flipped"): DPS is still the correct and default transport
/// for a single-mesh topology. What is being replaced is the ClusterClient branch, by gRPC.
/// </remarks>
private readonly IActorRef? _meshRouter;
private readonly ILoggingAdapter _log = Context.GetLogger();