diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs index 4c9e567e..34e0e3cb 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs @@ -48,6 +48,14 @@ public static class ServiceCollectionExtensions configuration, ConfigSourceOptions.SectionName); services.Configure(configuration.GetSection(ConfigServeOptions.SectionName)); + // Per-cluster mesh Phase 5: which transport carries a node's live-telemetry stream (node + // serve side + central dial side). Validated at startup for the same reason ConfigSource is — + // a Grpc-mode node with no listen port or no key produces silence, not an error. + services.AddValidatedOptions( + configuration, TelemetryOptions.SectionName); + services.AddValidatedOptions( + configuration, TelemetryDialOptions.SectionName); + services.AddSingleton(); return services; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs new file mode 100644 index 00000000..b9fc6d9a --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs @@ -0,0 +1,191 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.OtOpcUa.Cluster; + +/// +/// Node-side selection of the transport a node's live-telemetry stream is carried over +/// (per-cluster mesh Phase 5). The dark switch that lets a node stop publishing telemetry over +/// the mesh-wide DistributedPubSub topic and instead serve it over a dedicated gRPC stream that +/// does not require sharing a gossip ring with whoever is watching. +/// +public sealed class TelemetryOptions +{ + /// Configuration section name. + public const string SectionName = "Telemetry"; + + /// Publish telemetry over the mesh-wide DistributedPubSub topic — the default. + public const string ModeDps = "Dps"; + + /// Serve telemetry over a dedicated gRPC stream. + public const string ModeGrpc = "Grpc"; + + /// + /// (default) or . Any other value fails host + /// start. + /// + public string Mode { get; set; } = ModeDps; + + /// + /// Dedicated gRPC listen port for the telemetry stream. 0 (the default) disables it — + /// nothing is bound. Required on a driver-role node under . + /// + public int GrpcListenPort { get; set; } + + /// + /// Shared bearer key the serve-side interceptor checks. Supply via the environment + /// (Telemetry__ApiKey) — never commit it. Required under on any + /// roled node — fail-closed against hosting an un-keyed telemetry surface. + /// + public string ApiKey { get; set; } = string.Empty; +} + +/// +/// Central-side selection of how central dials a node for live telemetry (per-cluster mesh +/// Phase 5). Mirrors 's mode but carries the dial-side knobs +/// (contact refresh cadence, per-call timeout) instead of a listen port. +/// +public sealed class TelemetryDialOptions +{ + /// Configuration section name. + public const string SectionName = "TelemetryDial"; + + /// Read telemetry from the mesh-wide DistributedPubSub topic — the default. + public const string ModeDps = "Dps"; + + /// Dial nodes' dedicated gRPC telemetry streams. + public const string ModeGrpc = "Grpc"; + + /// + /// (default) or . Any other value fails host + /// start. + /// + public string Mode { get; set; } = ModeDps; + + /// + /// Shared bearer key; must equal a dialled node's . + /// Supply via the environment (TelemetryDial__ApiKey) — never commit it. Required + /// under . + /// + public string ApiKey { get; set; } = string.Empty; + + /// How often central refreshes its set of dialable node contacts, in seconds. + public int ContactRefreshSeconds { get; set; } = 60; + + /// Per-call deadline for a gRPC telemetry dial, in seconds. + public int CallTimeoutSeconds { get; set; } = 30; +} + +/// +/// Fails the host at startup on a shape that would leave a node +/// unable to serve its live-telemetry stream. +/// +/// +/// Like , every fault caught here otherwise surfaces as +/// an absence — a telemetry consumer that simply never sees data from this node, with no +/// stack trace and no failing node to point at; refusing to start is cheaper to diagnose. +/// +public sealed class TelemetryOptionsValidator : IValidateOptions +{ + private readonly IConfiguration _configuration; + + /// + /// DI-constructed by AddValidatedOptions (a plain AddSingleton), so a + /// constructor dependency is safe here. — not + /// IClusterRoleInfo — is the source of this node's roles: IClusterRoleInfo's + /// implementation needs the ActorSystem, which does not exist yet at + /// ValidateOnStart time. + /// + public TelemetryOptionsValidator(IConfiguration configuration) + { + _configuration = configuration; + } + + /// + public ValidateOptionsResult Validate(string? name, TelemetryOptions options) + { + ArgumentNullException.ThrowIfNull(options); + + var errors = new List(); + + var isDps = string.Equals(options.Mode, TelemetryOptions.ModeDps, StringComparison.OrdinalIgnoreCase); + var isGrpc = string.Equals(options.Mode, TelemetryOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); + + if (!isDps && !isGrpc) + { + errors.Add( + $"{TelemetryOptions.SectionName}:{nameof(TelemetryOptions.Mode)} is '{options.Mode}'. " + + $"Expected '{TelemetryOptions.ModeDps}' or '{TelemetryOptions.ModeGrpc}'."); + } + + var roles = _configuration.GetSection("Cluster:Roles").Get() ?? Array.Empty(); + var isDriver = Array.IndexOf(roles, "driver") >= 0; + + if (isGrpc) + { + if (isDriver && options.GrpcListenPort <= 0) + { + errors.Add( + $"Cluster:Roles has 'driver' and {TelemetryOptions.SectionName}:" + + $"{nameof(TelemetryOptions.Mode)} is '{TelemetryOptions.ModeGrpc}', but " + + $"{TelemetryOptions.SectionName}:{nameof(TelemetryOptions.GrpcListenPort)} is " + + $"{options.GrpcListenPort}. A driver node in Grpc telemetry mode must set " + + $"{TelemetryOptions.SectionName}:{nameof(TelemetryOptions.GrpcListenPort)}."); + } + + if (roles.Length > 0 && string.IsNullOrEmpty(options.ApiKey)) + { + errors.Add( + $"{TelemetryOptions.SectionName}:{nameof(TelemetryOptions.ApiKey)} is empty. Under " + + $"{TelemetryOptions.ModeGrpc} the shared bearer key is the whole authentication " + + "boundary to this node's telemetry surface; without it every dial is rejected. " + + $"Supply it via the environment ({TelemetryOptions.SectionName}__ApiKey)."); + } + } + + return errors.Count == 0 + ? ValidateOptionsResult.Success + : ValidateOptionsResult.Fail(string.Join(" ", errors)); + } +} + +/// +/// Fails the host at startup on a shape that would leave +/// central unable to dial a node's live-telemetry stream. +/// +public sealed class TelemetryDialOptionsValidator : IValidateOptions +{ + /// + public ValidateOptionsResult Validate(string? name, TelemetryDialOptions options) + { + ArgumentNullException.ThrowIfNull(options); + + var errors = new List(); + + var isDps = string.Equals( + options.Mode, TelemetryDialOptions.ModeDps, StringComparison.OrdinalIgnoreCase); + var isGrpc = string.Equals( + options.Mode, TelemetryDialOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); + + if (!isDps && !isGrpc) + { + errors.Add( + $"{TelemetryDialOptions.SectionName}:{nameof(TelemetryDialOptions.Mode)} is " + + $"'{options.Mode}'. Expected '{TelemetryDialOptions.ModeDps}' or " + + $"'{TelemetryDialOptions.ModeGrpc}'."); + } + + if (isGrpc && string.IsNullOrEmpty(options.ApiKey)) + { + errors.Add( + $"{TelemetryDialOptions.SectionName}:{nameof(TelemetryDialOptions.ApiKey)} is empty. " + + $"Under {TelemetryDialOptions.ModeGrpc} the shared bearer key is the whole " + + "authentication boundary to a node's telemetry surface; without it every dial is " + + $"rejected. Supply it via the environment ({TelemetryDialOptions.SectionName}__ApiKey)."); + } + + return errors.Count == 0 + ? ValidateOptionsResult.Success + : ValidateOptionsResult.Fail(string.Join(" ", errors)); + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/TelemetryOptionsValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/TelemetryOptionsValidatorTests.cs new file mode 100644 index 00000000..c55bd59a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/TelemetryOptionsValidatorTests.cs @@ -0,0 +1,157 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests; + +/// +/// A Grpc-mode node with no listen port or no key does not error — it simply never serves +/// telemetry, and whoever is watching sees nothing with no stack trace to point at. These tests +/// make that misconfiguration a loud host-start failure instead. +/// +public class TelemetryOptionsValidatorTests +{ + private static ValidateOptionsResult Validate(TelemetryOptions o, params string[] roles) + { + var pairs = new Dictionary(); + for (var i = 0; i < roles.Length; i++) + { + pairs[$"Cluster:Roles:{i}"] = roles[i]; + } + + var configuration = new ConfigurationBuilder().AddInMemoryCollection(pairs).Build(); + + return new TelemetryOptionsValidator(configuration).Validate(TelemetryOptions.SectionName, o); + } + + [Fact] + public void Default_options_are_valid() + { + Validate(new TelemetryOptions()).Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Unknown_mode_fails() + { + var result = Validate(new TelemetryOptions { Mode = "stream" }); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("stream"); + } + + [Fact] + public void Mode_matching_is_case_insensitive() + { + Validate(new TelemetryOptions + { + Mode = "grpc", + GrpcListenPort = 5100, + ApiKey = "k", + }, "driver").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_driver_node_with_no_listen_port_fails() + { + var result = Validate( + new TelemetryOptions { Mode = TelemetryOptions.ModeGrpc, ApiKey = "k" }, + "driver"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain(nameof(TelemetryOptions.GrpcListenPort)); + } + + [Fact] + public void Grpc_with_empty_key_fails() + { + var result = Validate( + new TelemetryOptions { Mode = TelemetryOptions.ModeGrpc, GrpcListenPort = 5100, ApiKey = "" }, + "driver"); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain(nameof(TelemetryOptions.ApiKey)); + } + + [Fact] + public void Grpc_driver_with_port_and_key_is_valid() + { + Validate( + new TelemetryOptions { Mode = TelemetryOptions.ModeGrpc, GrpcListenPort = 5100, ApiKey = "k" }, + "driver").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_admin_only_node_does_not_require_a_listen_port() + { + // The listen-port requirement is driver-specific — an admin-only node dialling out (via + // TelemetryDialOptions) never serves, so it has nothing to bind. + Validate( + new TelemetryOptions { Mode = TelemetryOptions.ModeGrpc, ApiKey = "k" }, + "admin").Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Grpc_with_no_roles_does_not_require_a_key() + { + // Fail-closed only applies to a node that actually carries a role — a roleless node hosts + // nothing to protect. + Validate(new TelemetryOptions { Mode = TelemetryOptions.ModeGrpc, ApiKey = "" }) + .Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Dps_ignores_the_empty_grpc_surface() + { + Validate( + new TelemetryOptions { Mode = TelemetryOptions.ModeDps, GrpcListenPort = 0, ApiKey = "" }, + "driver").Succeeded.ShouldBeTrue(); + } +} + +/// +/// Mirrors for the central dial-side options. +/// +public class TelemetryDialOptionsValidatorTests +{ + private static ValidateOptionsResult Validate(TelemetryDialOptions o) => + new TelemetryDialOptionsValidator().Validate(TelemetryDialOptions.SectionName, o); + + [Fact] + public void Default_options_are_valid() + { + Validate(new TelemetryDialOptions()).Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Unknown_mode_fails() + { + var result = Validate(new TelemetryDialOptions { Mode = "poll" }); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain("poll"); + } + + [Fact] + public void Grpc_with_empty_key_fails() + { + var result = Validate(new TelemetryDialOptions { Mode = TelemetryDialOptions.ModeGrpc, ApiKey = "" }); + + result.Failed.ShouldBeTrue(); + result.FailureMessage.ShouldContain(nameof(TelemetryDialOptions.ApiKey)); + } + + [Fact] + public void Grpc_with_a_key_is_valid() + { + Validate(new TelemetryDialOptions { Mode = TelemetryDialOptions.ModeGrpc, ApiKey = "k" }) + .Succeeded.ShouldBeTrue(); + } + + [Fact] + public void Dps_ignores_the_empty_key() + { + Validate(new TelemetryDialOptions { Mode = TelemetryDialOptions.ModeDps, ApiKey = "" }) + .Succeeded.ShouldBeTrue(); + } +}