Joseph Doherty
96 lines
4.3 KiB
C#
96 lines
4.3 KiB
C#
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
|
|
|
|
/// <summary>
|
|
/// Production seams for <see cref="TelemetryDialSupervisor"/>: the DB-backed node source and the
|
|
/// gRPC dial loop. Kept out of the actor so the actor stays DB-free and gRPC-free (and therefore
|
|
/// unit-testable with fakes).
|
|
/// </summary>
|
|
public static class TelemetryNodeSource
|
|
{
|
|
/// <summary>
|
|
/// Builds the node source the supervisor polls: enabled, non-maintenance <c>ClusterNode</c>
|
|
/// rows carrying a non-null <c>GrpcPort</c>, projected to <c>http://{Host}:{GrpcPort}</c>
|
|
/// h2c endpoints. A row with a null <c>GrpcPort</c> exposes no telemetry surface yet and is
|
|
/// skipped with a Warning — mirrors <c>CentralCommunicationActor.LoadContactsFromDb</c>'s
|
|
/// enabled + non-maintenance read.
|
|
/// </summary>
|
|
/// <param name="dbFactory">Factory for the config database holding <c>ClusterNode</c> rows.</param>
|
|
/// <param name="logger">Logger for skipped-row diagnostics.</param>
|
|
/// <returns>An async delegate resolving the current dialable node set.</returns>
|
|
public static Func<Task<IReadOnlyList<TelemetryDialTarget>>> Create(
|
|
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
|
ILogger logger)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(dbFactory);
|
|
ArgumentNullException.ThrowIfNull(logger);
|
|
|
|
// A persistently null-GrpcPort node is skipped on every refresh (every ContactRefreshSeconds);
|
|
// warn once per node id so it does not spam the log forever.
|
|
var warnedNullPort = new HashSet<string>(StringComparer.Ordinal);
|
|
|
|
return async () =>
|
|
{
|
|
await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false);
|
|
var rows = await db.ClusterNodes
|
|
.AsNoTracking()
|
|
.Where(n => n.Enabled && !n.MaintenanceMode)
|
|
.Select(n => new { n.NodeId, n.Host, n.GrpcPort })
|
|
.ToListAsync()
|
|
.ConfigureAwait(false);
|
|
|
|
var targets = new List<TelemetryDialTarget>(rows.Count);
|
|
foreach (var row in rows)
|
|
{
|
|
if (row.GrpcPort is null)
|
|
{
|
|
if (warnedNullPort.Add(row.NodeId))
|
|
{
|
|
logger.LogWarning(
|
|
"ClusterNode {NodeId} has no GrpcPort; it exposes no telemetry stream and is "
|
|
+ "skipped in this dial refresh (further skips of this node are silent)",
|
|
row.NodeId);
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
targets.Add(new TelemetryDialTarget(row.NodeId, $"http://{row.Host}:{row.GrpcPort}"));
|
|
}
|
|
|
|
return (IReadOnlyList<TelemetryDialTarget>)targets;
|
|
};
|
|
}
|
|
|
|
/// <summary>
|
|
/// Builds the production dial loop: one <see cref="TelemetryStreamClient"/> per stream, mapping
|
|
/// every wire envelope via <see cref="TelemetryProtoMapCentral.MapEvent"/> before handing it to
|
|
/// the supervisor. The client isolates a mapper throw (logs + continues) and routes only
|
|
/// transport failures to <c>onError</c>, so the supervisor sees a poison event as "dropped",
|
|
/// never as "node down".
|
|
/// </summary>
|
|
/// <param name="apiKey">Shared node bearer key sent on every dial.</param>
|
|
/// <param name="clientLogger">Logger passed to each per-stream client (dropped-event diagnostics).</param>
|
|
/// <returns>The dial-loop delegate the supervisor invokes per generation.</returns>
|
|
public static TelemetryDialLoop CreateDialLoop(
|
|
string apiKey,
|
|
ILogger<TelemetryStreamClient>? clientLogger)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(apiKey);
|
|
|
|
return async (target, correlationId, onMapped, onError, ct) =>
|
|
{
|
|
using var client = new TelemetryStreamClient(target.Endpoint, apiKey, clientLogger);
|
|
await client.RunAsync(
|
|
correlationId,
|
|
onEvent: evt => onMapped(TelemetryProtoMapCentral.MapEvent(evt)),
|
|
onError: onError,
|
|
ct).ConfigureAwait(false);
|
|
};
|
|
}
|
|
}
|