feat(mesh-phase5): Telemetry/TelemetryDial options + fail-closed validators
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -48,6 +48,14 @@ public static class ServiceCollectionExtensions
|
||||
configuration, ConfigSourceOptions.SectionName);
|
||||
services.Configure<ConfigServeOptions>(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<TelemetryOptions, TelemetryOptionsValidator>(
|
||||
configuration, TelemetryOptions.SectionName);
|
||||
services.AddValidatedOptions<TelemetryDialOptions, TelemetryDialOptionsValidator>(
|
||||
configuration, TelemetryDialOptions.SectionName);
|
||||
|
||||
services.AddSingleton<IClusterRoleInfo, ClusterRoleInfo>();
|
||||
|
||||
return services;
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Cluster;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class TelemetryOptions
|
||||
{
|
||||
/// <summary>Configuration section name.</summary>
|
||||
public const string SectionName = "Telemetry";
|
||||
|
||||
/// <summary>Publish telemetry over the mesh-wide DistributedPubSub topic — the default.</summary>
|
||||
public const string ModeDps = "Dps";
|
||||
|
||||
/// <summary>Serve telemetry over a dedicated gRPC stream.</summary>
|
||||
public const string ModeGrpc = "Grpc";
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="ModeDps"/> (default) or <see cref="ModeGrpc"/>. Any other value fails host
|
||||
/// start.
|
||||
/// </summary>
|
||||
public string Mode { get; set; } = ModeDps;
|
||||
|
||||
/// <summary>
|
||||
/// Dedicated gRPC listen port for the telemetry stream. <c>0</c> (the default) disables it —
|
||||
/// nothing is bound. Required on a driver-role node under <see cref="ModeGrpc"/>.
|
||||
/// </summary>
|
||||
public int GrpcListenPort { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Shared bearer key the serve-side interceptor checks. Supply via the environment
|
||||
/// (<c>Telemetry__ApiKey</c>) — never commit it. Required under <see cref="ModeGrpc"/> on any
|
||||
/// roled node — fail-closed against hosting an un-keyed telemetry surface.
|
||||
/// </summary>
|
||||
public string ApiKey { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Central-side selection of how central dials a node for live telemetry (per-cluster mesh
|
||||
/// Phase 5). Mirrors <see cref="TelemetryOptions"/>'s mode but carries the dial-side knobs
|
||||
/// (contact refresh cadence, per-call timeout) instead of a listen port.
|
||||
/// </summary>
|
||||
public sealed class TelemetryDialOptions
|
||||
{
|
||||
/// <summary>Configuration section name.</summary>
|
||||
public const string SectionName = "TelemetryDial";
|
||||
|
||||
/// <summary>Read telemetry from the mesh-wide DistributedPubSub topic — the default.</summary>
|
||||
public const string ModeDps = "Dps";
|
||||
|
||||
/// <summary>Dial nodes' dedicated gRPC telemetry streams.</summary>
|
||||
public const string ModeGrpc = "Grpc";
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="ModeDps"/> (default) or <see cref="ModeGrpc"/>. Any other value fails host
|
||||
/// start.
|
||||
/// </summary>
|
||||
public string Mode { get; set; } = ModeDps;
|
||||
|
||||
/// <summary>
|
||||
/// Shared bearer key; must equal a dialled node's <see cref="TelemetryOptions.ApiKey"/>.
|
||||
/// Supply via the environment (<c>TelemetryDial__ApiKey</c>) — never commit it. Required
|
||||
/// under <see cref="ModeGrpc"/>.
|
||||
/// </summary>
|
||||
public string ApiKey { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>How often central refreshes its set of dialable node contacts, in seconds.</summary>
|
||||
public int ContactRefreshSeconds { get; set; } = 60;
|
||||
|
||||
/// <summary>Per-call deadline for a gRPC telemetry dial, in seconds.</summary>
|
||||
public int CallTimeoutSeconds { get; set; } = 30;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fails the host at startup on a <see cref="TelemetryOptions"/> shape that would leave a node
|
||||
/// unable to serve its live-telemetry stream.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Like <see cref="ConfigSourceOptionsValidator"/>, every fault caught here otherwise surfaces as
|
||||
/// an <i>absence</i> — 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.
|
||||
/// </remarks>
|
||||
public sealed class TelemetryOptionsValidator : IValidateOptions<TelemetryOptions>
|
||||
{
|
||||
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.
|
||||
/// </summary>
|
||||
public TelemetryOptionsValidator(IConfiguration configuration)
|
||||
{
|
||||
_configuration = configuration;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValidateOptionsResult Validate(string? name, TelemetryOptions options)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
|
||||
var errors = new List<string>();
|
||||
|
||||
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<string[]>() ?? Array.Empty<string>();
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fails the host at startup on a <see cref="TelemetryDialOptions"/> shape that would leave
|
||||
/// central unable to dial a node's live-telemetry stream.
|
||||
/// </summary>
|
||||
public sealed class TelemetryDialOptionsValidator : IValidateOptions<TelemetryDialOptions>
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public ValidateOptionsResult Validate(string? name, TelemetryDialOptions options)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
|
||||
var errors = new List<string>();
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public class TelemetryOptionsValidatorTests
|
||||
{
|
||||
private static ValidateOptionsResult Validate(TelemetryOptions o, params string[] roles)
|
||||
{
|
||||
var pairs = new Dictionary<string, string?>();
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Mirrors <see cref="TelemetryOptionsValidatorTests"/> for the central dial-side options.
|
||||
/// </summary>
|
||||
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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user