fix(host): make a failed OPC UA SDK start visible and stop the SDK on StopAsync
A failed SDK start (port in use, PKI unwritable, bad OpcUa:* override) was one Error line: the
node stayed /health/ready 200, joined its mesh, could be elected driver Primary and acked deploys
through the Null sink while every client got connection-refused.
- IOpcUaEndpointState / OpcUaEndpointState: the start outcome is a fact the hosted service
maintains (NotStarted / Started / Failed / Stopped + reason).
- OpcUaEndpointHealthCheck ("opcua-endpoint", Ready + Active) is Unhealthy unless Started;
registered by AddOtOpcUaHealth(hasAdmin, hasDriver) on driver-role nodes.
- A driver-only node rethrows the start failure (Program.cs's fatal handler exits non-zero so
the supervisor restarts it); a fused node keeps its AdminUI up but is visibly not ready.
- StopAsync detaches the sinks and then STOPS the SDK (the port closes, clients get
connection-refused rather than a stale, inert endpoint); DisposeAsync is idempotent with it.
- The ServiceLevel calculator / redundancy election live in Cluster + Runtime (not this lane);
the fact is exposed for the Wave 3 follow-up to consume.
Tests: port-in-use ⇒ state Failed, probe Unhealthy, real /health/ready and /health/active 503
(positive control: 200 once Started); driver-only rethrows; StopAsync closes the port and
DisposeAsync twice is a no-op; the probe registration is on both tiers for driver nodes only.
Refs #733
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -49,6 +49,34 @@ public sealed class ActiveTierRegistrationTests
|
||||
ClusterPrimaryRegistration(hasAdmin).Tags.ShouldContain(ZbHealthTags.Active);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// R03:2.1 — the OPC UA endpoint probe is on BOTH tiers of a driver node (it cannot serve, and must not
|
||||
/// be routed to) and absent from an admin-only node, which starts no endpoint to be degraded by.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(true)]
|
||||
[InlineData(false)]
|
||||
public void Opcua_endpoint_probe_is_registered_on_ready_and_active_for_driver_nodes_only(bool hasDriver)
|
||||
{
|
||||
var registrations = new ServiceCollection()
|
||||
.AddOtOpcUaHealth(hasAdmin: true, hasDriver: hasDriver)
|
||||
.BuildServiceProvider()
|
||||
.GetRequiredService<IOptions<HealthCheckServiceOptions>>()
|
||||
.Value.Registrations
|
||||
.Where(r => r.Name == OpcUaEndpointHealthCheck.Name)
|
||||
.ToList();
|
||||
|
||||
if (!hasDriver)
|
||||
{
|
||||
registrations.ShouldBeEmpty();
|
||||
return;
|
||||
}
|
||||
|
||||
var registration = registrations.ShouldHaveSingleItem();
|
||||
registration.Tags.ShouldContain(ZbHealthTags.Ready);
|
||||
registration.Tags.ShouldContain(ZbHealthTags.Active);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Active_tier_check_is_not_on_the_ready_tier()
|
||||
{
|
||||
|
||||
+213
@@ -0,0 +1,213 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using Akka.Hosting;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Diagnostics.HealthChecks;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Cluster;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Health;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
/// <summary>
|
||||
/// R03:2.1 / 1.4 — a failed OPC UA SDK start used to be one Error line: the node stayed Ready, joined
|
||||
/// its mesh, and acked deploys through the Null sink while clients got connection-refused. The start
|
||||
/// outcome is now a fact (<see cref="IOpcUaEndpointState"/>) the ready/active tiers read, a driver-only
|
||||
/// node rethrows so the process exits, and <c>StopAsync</c> actually stops the SDK.
|
||||
/// </summary>
|
||||
public sealed class OtOpcUaServerHostedServiceStartFailureTests
|
||||
{
|
||||
private static readonly TimeSpan Deadline = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// Port already bound ⇒ the SDK cannot listen ⇒ the state is Failed, the probe is Unhealthy and the
|
||||
/// real <c>/health/ready</c> endpoint answers 503 — while StartAsync itself returns (fused node).
|
||||
/// Positive control: the same endpoint answers 200 once the state is Started.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Port_in_use_marks_the_endpoint_failed_and_health_ready_answers_503()
|
||||
{
|
||||
using var deadline = new CancellationTokenSource(Deadline);
|
||||
using var squatter = SquatAllInterfaces(out var port);
|
||||
var pki = NewPki();
|
||||
var state = new OpcUaEndpointState();
|
||||
var fused = EffectiveRoles.Resolve(new[] { "admin", "driver" }, envRoles: null);
|
||||
|
||||
var service = NewService(port, pki, state, fused);
|
||||
try
|
||||
{
|
||||
await service.StartAsync(deadline.Token); // must NOT throw on a fused node
|
||||
|
||||
state.Status.ShouldBe(OpcUaEndpointStatus.Failed);
|
||||
state.FailureReason.ShouldNotBeNullOrWhiteSpace();
|
||||
|
||||
var probe = await new OpcUaEndpointHealthCheck(state).CheckHealthAsync(new HealthCheckContext());
|
||||
probe.Status.ShouldBe(HealthStatus.Unhealthy);
|
||||
probe.Description.ShouldNotBeNull();
|
||||
probe.Description.ShouldContain("failed to start");
|
||||
|
||||
await using var web = BuildHealthHost(state);
|
||||
await web.StartAsync(deadline.Token);
|
||||
using var http = new HttpClient { BaseAddress = new Uri(web.Urls.First()) };
|
||||
|
||||
(await http.GetAsync("/health/ready", deadline.Token)).StatusCode.ShouldBe(HttpStatusCode.ServiceUnavailable);
|
||||
(await http.GetAsync("/health/active", deadline.Token)).StatusCode.ShouldBe(HttpStatusCode.ServiceUnavailable);
|
||||
|
||||
// Positive control on the endpoint: the same host answers 200 for a started endpoint, so the
|
||||
// 503 above measured the probe and not a broken health pipeline.
|
||||
state.MarkStarted();
|
||||
(await http.GetAsync("/health/ready", deadline.Token)).StatusCode.ShouldBe(HttpStatusCode.OK);
|
||||
await web.StopAsync(deadline.Token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.DisposeAsync();
|
||||
squatter.Stop();
|
||||
Cleanup(pki);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A driver-only node has nothing else to stay up for: the failed start propagates.</summary>
|
||||
[Fact]
|
||||
public async Task Driver_only_node_rethrows_when_the_sdk_fails_to_start()
|
||||
{
|
||||
using var deadline = new CancellationTokenSource(Deadline);
|
||||
using var squatter = SquatAllInterfaces(out var port);
|
||||
var pki = NewPki();
|
||||
var state = new OpcUaEndpointState();
|
||||
var driverOnly = EffectiveRoles.Resolve(new[] { "driver" }, envRoles: null);
|
||||
|
||||
var service = NewService(port, pki, state, driverOnly);
|
||||
try
|
||||
{
|
||||
await Should.ThrowAsync<Exception>(() => service.StartAsync(deadline.Token));
|
||||
state.Status.ShouldBe(OpcUaEndpointStatus.Failed);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.DisposeAsync();
|
||||
squatter.Stop();
|
||||
Cleanup(pki);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A clean start is Started and listening; <c>StopAsync</c> closes the port (connection refused, not
|
||||
/// a stale-but-listening endpoint) and marks Stopped; <c>DisposeAsync</c> twice afterwards is a no-op.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StopAsync_closes_the_port_and_DisposeAsync_is_idempotent()
|
||||
{
|
||||
using var deadline = new CancellationTokenSource(Deadline);
|
||||
var port = FreePort();
|
||||
var pki = NewPki();
|
||||
var state = new OpcUaEndpointState();
|
||||
var service = NewService(port, pki, state, EffectiveRoles.Resolve(new[] { "driver" }, envRoles: null));
|
||||
try
|
||||
{
|
||||
await service.StartAsync(deadline.Token);
|
||||
state.Status.ShouldBe(OpcUaEndpointStatus.Started);
|
||||
using (var client = new TcpClient())
|
||||
{
|
||||
await client.ConnectAsync(IPAddress.Loopback, port, deadline.Token);
|
||||
client.Connected.ShouldBeTrue("positive control: the endpoint is listening after StartAsync");
|
||||
}
|
||||
|
||||
await service.StopAsync(deadline.Token);
|
||||
|
||||
state.Status.ShouldBe(OpcUaEndpointStatus.Stopped);
|
||||
using (var client = new TcpClient())
|
||||
{
|
||||
await Should.ThrowAsync<SocketException>(async () => await client.ConnectAsync(IPAddress.Loopback, port, deadline.Token));
|
||||
}
|
||||
|
||||
await service.DisposeAsync();
|
||||
await service.DisposeAsync(); // idempotent
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.DisposeAsync();
|
||||
Cleanup(pki);
|
||||
}
|
||||
}
|
||||
|
||||
private static OtOpcUaServerHostedService NewService(int port, string pki, OpcUaEndpointState state, EffectiveRoles roles) =>
|
||||
new(
|
||||
Options.Create(new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.StartFailure",
|
||||
ApplicationUri = $"urn:OtOpcUa.StartFailure:{Guid.NewGuid():N}",
|
||||
OpcUaPort = port, PublicHostname = "localhost", PkiStoreRoot = pki,
|
||||
}),
|
||||
new DeferredAddressSpaceSink(), new DeferredServiceLevelPublisher(),
|
||||
NullOpcUaUserAuthenticator.Instance, NullHistorianDataSource.Instance,
|
||||
EmptyDataPlaneAuthoritySource.Instance, new ConfigurationBuilder().Build(),
|
||||
() => throw new NotSupportedException("no actor system required"), new ActorRegistry(),
|
||||
null, Options.Create(new AclOptions()), TimeProvider.System, NullLoggerFactory.Instance,
|
||||
endpointState: state, roles: roles);
|
||||
|
||||
/// <summary>The production health registration (driver role) over the supplied state, on a loopback Kestrel.</summary>
|
||||
private static WebApplication BuildHealthHost(OpcUaEndpointState state)
|
||||
{
|
||||
var builder = WebApplication.CreateBuilder();
|
||||
builder.WebHost.UseUrls("http://127.0.0.1:0");
|
||||
builder.Logging.ClearProviders();
|
||||
builder.Services.AddSingleton<IOpcUaEndpointState>(state);
|
||||
builder.Services.AddOtOpcUaHealth(hasAdmin: false, hasDriver: true);
|
||||
// The akka + cluster-primary probes need a running ActorSystem; keep only the probe under test so
|
||||
// the status measured is the endpoint's. (The registration shape itself is pinned separately.)
|
||||
builder.Services.Configure<HealthCheckServiceOptions>(o =>
|
||||
{
|
||||
foreach (var r in o.Registrations.Where(r => r.Name != OpcUaEndpointHealthCheck.Name).ToList())
|
||||
o.Registrations.Remove(r);
|
||||
});
|
||||
var app = builder.Build();
|
||||
app.MapOtOpcUaHealth();
|
||||
return app;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Holds a port on the IPv4 AND IPv6 wildcard addresses. The SDK binds its listener on the
|
||||
/// wildcards, and on macOS a loopback-only squatter does not conflict with a wildcard bind (both
|
||||
/// sockets carry SO_REUSEADDR), so a loopback squatter would let the SDK start — the first run of
|
||||
/// this test proved exactly that. The SDK also only fails when NEITHER family binds.
|
||||
/// </summary>
|
||||
private static TcpListener SquatAllInterfaces(out int port)
|
||||
{
|
||||
var squatter = new TcpListener(IPAddress.IPv6Any, 0);
|
||||
squatter.Server.DualMode = true;
|
||||
squatter.ExclusiveAddressUse = true;
|
||||
squatter.Start();
|
||||
port = ((IPEndPoint)squatter.LocalEndpoint).Port;
|
||||
return squatter;
|
||||
}
|
||||
|
||||
private static int FreePort()
|
||||
{
|
||||
using var probe = new TcpListener(IPAddress.Loopback, 0);
|
||||
probe.Start();
|
||||
var port = ((IPEndPoint)probe.LocalEndpoint).Port;
|
||||
probe.Stop();
|
||||
return port;
|
||||
}
|
||||
|
||||
private static string NewPki() => Path.Combine(Path.GetTempPath(), $"otopcua-startfail-{Guid.NewGuid():N}");
|
||||
|
||||
private static void Cleanup(string pki)
|
||||
{
|
||||
if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user