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

test(mesh-phase5): 3-port listener reachability + comment fix for the telemetry endpoint

Browse Source 
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-23 16:18:15 -04:00
parent a279a43ed4
commit af9c9d78da
3 changed files with 195 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs
+14 -8
View File
@@ -417,21 +417,25 @@ builder.Services.AddOtOpcUaSecrets(builder.Configuration);
builder.Services.AddOtOpcUaHealth(hasAdmin);
builder.Services.AddOtOpcUaObservability(builder.Configuration);
// gRPC server plumbing shared by two endpoints: the LocalDb passive sync endpoint (driver-role,
// mapped below on hasDriver && syncListenPort > 0) and the ConfigServe artifact endpoint
// (admin-role, mapped on hasAdmin && configServeGrpcPort > 0). Each interceptor is scoped strictly
// to its own service path, so both can share one pipeline and each is a harmless pass-through when
// its service is unmapped. Both fail-closed with no ApiKey configured — the interceptor is the ONLY
// inbound auth on either endpoint (the libraries verify nothing). Registered whenever either role is
// present so an admin-only node still serves config and a driver-only node still gates sync.
// gRPC server plumbing shared by three endpoints: the LocalDb passive sync endpoint (driver-role,
// mapped below on hasDriver && syncListenPort > 0), the ConfigServe artifact endpoint (admin-role,
// mapped on hasAdmin && configServeGrpcPort > 0), and the telemetry-stream endpoint (driver-role,
// mapped on hasDriver && telemetryListenPort > 0, per-cluster mesh Phase 5). Each interceptor is
// scoped strictly to its own service path, so all three can share one pipeline and each is a
// harmless pass-through when its service is unmapped. All three fail-closed with no ApiKey
// configured (LocalDbSyncAuthInterceptor, ConfigServeAuthInterceptor, TelemetryStreamAuthInterceptor)
// — the interceptor is the ONLY inbound auth on any endpoint (the libraries verify nothing).
// Registered whenever either role is present so an admin-only node still serves config and a
// driver-only node still gates sync + telemetry.
if (hasDriver || hasAdmin)
{
builder.Services.AddGrpc(o =>
{
if (hasDriver)
{
o.Interceptors.Add<LocalDbSyncAuthInterceptor>();
if (hasDriver)
o.Interceptors.Add<TelemetryStreamAuthInterceptor>();
}
if (hasAdmin)
o.Interceptors.Add<ConfigServeAuthInterceptor>();
});
@@ -596,7 +600,9 @@ if (hasAdmin && configServeGrpcPort > 0)
// Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by
// TelemetryStreamAuthInterceptor (fail-closed on empty Telemetry:ApiKey).
if (hasDriver && telemetryListenPort > 0)
{
app.MapGrpcService<TelemetryStreamGrpcService>();
}
app.MapOtOpcUaHealth();
app.MapOtOpcUaMetrics();
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TelemetryListenerTests.cs
+176
View File
@@ -0,0 +1,176 @@
using System.Net;
using System.Threading.Channels;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Moq;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.Grpc;
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;
/// <summary>
/// Per-cluster mesh Phase 5 (host-wiring task) — the telemetry-stream h2c listener coexisting with
/// the two earlier dedicated ports (LocalDb sync + ConfigServe) and the main HTTP surface, and the
/// <see cref="TelemetryStreamGrpcService"/> actually being <b>mapped and interceptor-gated</b> on
/// that port.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why this exists.</b> Phase 5 adds a THIRD dedicated h2c port to <c>Program.cs</c>'s merged
/// listener block. That block must still re-bind the existing HTTP surface EXACTLY ONCE while
/// adding all three dedicated ports (re-binding it per-port throws "address already in use";
/// re-binding it zero times silently unbinds the AdminUI behind Traefik). The sibling
/// <see cref="ConfigServeListenerTests"/> pins the two-port case; this pins the three-port case
/// AND — the part a port-open check cannot prove — that the telemetry gRPC service is genuinely
/// wired: an un-keyed dial gets <see cref="StatusCode.PermissionDenied"/> (mapped + interceptor
/// reached), not <see cref="StatusCode.Unimplemented"/> (never mapped), and a correctly-keyed
/// dial opens a real stream. This is the executable regression for the "ships clean but never
/// actually wired" gap.
/// </para>
/// <para>
/// Like <see cref="ConfigServeListenerTests"/>, this drives a minimal <c>WebApplication</c>
/// shaped exactly like <c>Program.cs</c>'s driver-role wiring (AddGrpc +
/// <see cref="TelemetryStreamAuthInterceptor"/> + <c>Configure&lt;TelemetryOptions&gt;</c> +
/// <c>MapGrpcService&lt;TelemetryStreamGrpcService&gt;</c>) rather than booting the real host
/// (which needs SQL, an Akka mesh and LDAP). What is under test is the Kestrel binding contract
/// and the gRPC map/intercept wiring, both fully reproduced here. The node's own telemetry hub
/// is a stub — this test proves reachability, not fan-out content.
/// </para>
/// </remarks>
public sealed class TelemetryListenerTests
{
private const string ApiKey = "telemetry-test-key";
[Fact]
public async Task AllThreeDedicatedH2cPorts_AndTheHttpSurface_AnswerSimultaneously_AndTelemetryServiceIsWired()
{
var httpPort = GetFreePort();
var syncPort = GetFreePort();
var configServePort = GetFreePort();
var telemetryPort = GetFreePort();
var builder = WebApplication.CreateBuilder();
builder.Environment.ApplicationName = typeof(TelemetryListenerTests).Assembly.GetName().Name!;
// Exactly the driver-role wiring Program.cs applies for the telemetry endpoint: the fail-closed
// interceptor on the shared AddGrpc pipeline, the TelemetryOptions the interceptor reads its
// expected key from, and a stub hub so the mapped service can be constructed on the keyed path.
builder.Services.Configure<TelemetryOptions>(o => o.ApiKey = ApiKey);
builder.Services.AddSingleton(StubHubYieldingAnEmptyStream());
builder.Services.AddGrpc(o => o.Interceptors.Add<TelemetryStreamAuthInterceptor>());
// The shape Program.cs's merged block applies: compute the existing surface ONCE, apply it ONCE,
// then add each configured dedicated h2c port. Here all THREE dedicated ports are set at once —
// the fused-node case that would double-bind the HTTP surface if the block re-applied it per-port.
var existingBindings = KestrelHttpBinding.Parse($"http://localhost:{httpPort}");
builder.WebHost.ConfigureKestrel(kestrel =>
{
foreach (var binding in existingBindings)
binding.Apply(kestrel);
kestrel.ListenAnyIP(syncPort, o => o.Protocols = HttpProtocols.Http2);
kestrel.ListenAnyIP(configServePort, o => o.Protocols = HttpProtocols.Http2);
kestrel.ListenAnyIP(telemetryPort, o => o.Protocols = HttpProtocols.Http2);
});
await using var app = builder.Build();
app.MapGet("/healthz", () => "ok");
app.MapGrpcService<TelemetryStreamGrpcService>();
await app.StartAsync(TestContext.Current.CancellationToken);
try
{
var ct = TestContext.Current.CancellationToken;
// 1) The re-bound HTTP/1.1 surface still answers (the AdminUI/Traefik guarantee) even with
// all three dedicated ports added — i.e. the existing surface was re-bound exactly once.
using var http1 = new HttpClient();
(await http1.GetStringAsync($"http://localhost:{httpPort}/healthz", ct)).ShouldBe("ok");
// 2) The sync + config-serve dedicated ports negotiate prior-knowledge h2c. A 404 is fine —
// it proves the HTTP/2 connection established and routed, which is all that is in question
// for those two (their services are mapped elsewhere; here we only pin the port binding).
using var http2 = new HttpClient
{
DefaultRequestVersion = HttpVersion.Version20,
DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact,
};
(await http2.GetAsync($"http://localhost:{syncPort}/", ct)).Version.ShouldBe(HttpVersion.Version20);
(await http2.GetAsync($"http://localhost:{configServePort}/", ct)).Version.ShouldBe(HttpVersion.Version20);
// 3) The telemetry port serves the MAPPED, interceptor-gated gRPC service. Dial it for real.
using var channel = GrpcChannel.ForAddress(
$"http://localhost:{telemetryPort}", new GrpcChannelOptions { HttpHandler = new SocketsHttpHandler() });
var client = new TelemetryStreamService.TelemetryStreamServiceClient(channel);
// 3a) No bearer -> PermissionDenied. This is the load-bearing assertion: PermissionDenied can
// only come from the interceptor, and the interceptor only runs because the method was
// routed to the mapped service. A service that was never mapped would answer Unimplemented.
var noKey = await Should.ThrowAsync<RpcException>(async () =>
{
using var call = client.Subscribe(new TelemetryStreamRequest { CorrelationId = "no-key" }, cancellationToken: ct);
await foreach (var _ in call.ResponseStream.ReadAllAsync(ct)) { }
});
noKey.StatusCode.ShouldBe(StatusCode.PermissionDenied);
noKey.StatusCode.ShouldNotBe(StatusCode.Unimplemented);
// 3b) Correct bearer -> the mapped handler actually executes end-to-end (interceptor passes,
// the service resolves the stub hub and streams). The stub yields an empty, completed
// stream, so the call completes cleanly with zero envelopes.
var headers = new Metadata { { "authorization", $"Bearer {ApiKey}" } };
using var keyed = client.Subscribe(
new TelemetryStreamRequest { CorrelationId = "keyed" }, headers, cancellationToken: ct);
var received = 0;
await foreach (var _ in keyed.ResponseStream.ReadAllAsync(ct))
received++;
received.ShouldBe(0);
}
finally
{
await app.StopAsync(TestContext.Current.CancellationToken);
}
}
/// <summary>
/// A hub whose subscription reader is already completed and empty, so the keyed
/// <c>Subscribe</c> handler runs to completion and closes the stream with zero envelopes —
/// enough to prove the mapped service executed without depending on any live telemetry producer.
/// </summary>
private static ITelemetryLocalHub StubHubYieldingAnEmptyStream()
{
var hub = new Mock<ITelemetryLocalHub>();
hub.Setup(h => h.Subscribe(It.IsAny<int>())).Returns(() => new EmptySubscription());
return hub.Object;
}
private sealed class EmptySubscription : ITelemetrySubscription
{
private readonly Channel<TelemetryItem> _channel = Channel.CreateBounded<TelemetryItem>(1);
public EmptySubscription() => _channel.Writer.Complete();
public ChannelReader<TelemetryItem> Reader => _channel.Reader;
public void Dispose()
{
}
}
private static int GetFreePort()
{
using var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0);
listener.Start();
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
listener.Stop();
return port;
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.csproj
+5
View File
@@ -9,6 +9,11 @@
<ItemGroup>
<PackageReference Include="Moq" />
<!-- Grpc.Net.Client: the initiator side of a gRPC call. The Host references only
Grpc.AspNetCore (server side); TelemetryListenerTests dials the mapped telemetry service
to prove it is reachable, which needs the client. Grpc.Core.Api (RpcException/StatusCode/
Metadata + the generated client base) flows in transitively via the Commons project ref. -->
<PackageReference Include="Grpc.Net.Client" />
<PackageReference Include="xunit.v3" />
<PackageReference Include="Shouldly" />
<PackageReference Include="Microsoft.NET.Test.Sdk" />