Files
lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs
T
Joseph Doherty c6d4b14c93 feat(telemetry): wire TelemetryDial:CallTimeoutSeconds as the dial-establishment deadline (A5)
CallTimeoutSeconds was bound, defaulted and documented but consumed
nowhere. TelemetryDialSupervisor now arms a per-generation deadline on
every dial: a dial that has delivered no event by the deadline is
cancelled and treated as a transport failure (reconnect with a failure
streak). A live node replays its snapshots on subscribe, so this bites
only a dial that never connects (a black-holed host, where the socket
connect has no timeout). An established stream is deliberately not
bounded — the wire has no heartbeat, keepalive pings own its liveness.

Review ref: 02/U2.

Refs #731

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:38:32 -04:00

207 lines
9.5 KiB
C#

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>
/// Deadline, in seconds, for a gRPC telemetry dial to deliver its first event. A live node answers a
/// fresh subscribe at once (it replays its health/resilience snapshots), so a dial still silent at the
/// deadline is one that never connected — it is cancelled and re-dialled. Bounds establishment only;
/// an established stream is kept alive by HTTP/2 keepalive pings, not by this value. Values below 1
/// are treated as 1.
/// </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 EffectiveRoles _effectiveRoles;
/// <summary>
/// DI-constructed by <c>AddValidatedOptions</c> (a plain <c>AddSingleton</c>), so a
/// constructor dependency is safe here. This validator's only use of configuration is this
/// node's roles, so it takes <see cref="EffectiveRoles"/> — the single role authority
/// — directly rather than <see cref="IConfiguration"/>.
/// </summary>
/// <param name="effectiveRoles">The node's resolved roles, the single role authority.</param>
public TelemetryOptionsValidator(EffectiveRoles effectiveRoles)
{
_effectiveRoles = effectiveRoles;
}
/// <summary>Validates the telemetry options against this node's roles.</summary>
/// <param name="name">The named options instance, or <see langword="null"/> for the default.</param>
/// <param name="options">The options to validate.</param>
/// <returns>Success, or a failure listing every invalid setting.</returns>
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}'.");
}
// EffectiveRoles is the single role authority — its precedence already
// applies the OTOPCUA_ROLES fallback and its normalization already folds case, so
// a node configured only through the env var, or spelled 'Driver', still satisfies the
// Grpc-mode requirements below.
var isDriver = _effectiveRoles.HasDriver;
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 (_effectiveRoles.Gossiped.Count > 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>
{
/// <summary>Validates the telemetry dial options.</summary>
/// <param name="name">The named options instance, or <see langword="null"/> for the default.</param>
/// <param name="options">The options to validate.</param>
/// <returns>Success, or a failure listing every invalid setting.</returns>
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));
}
}