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

feat(batch7): implement f1 config receiver and reload helpers

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-28 10:53:25 -05:00
parent 996c552d2f
commit 3908ecdcb1
3 changed files with 731 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

398
dotnet/src/ZB.MOM.NatsNet.Server/Config/ReloadOptions.cs
View File

@@ -13,6 +13,11 @@
//
// Adapted from server/reload.go in the NATS server Go source.
using System.Reflection;
using System.Net.Security;
using ZB.MOM.NatsNet.Server.Auth;
using ZB.MOM.NatsNet.Server.Internal;
namespace ZB.MOM.NatsNet.Server;
// =============================================================================
@@ -74,7 +79,7 @@ public interface IReloadOption
public abstract class NoopReloadOption : IReloadOption
{
/// <inheritdoc/>
public virtual void Apply(NatsServer server) { }
public virtual void Apply(NatsServer server) => _ = server;
/// <inheritdoc/>
public virtual bool IsLoggingChange() => false;
@@ -196,7 +201,7 @@ internal sealed class DebugReloadOption : LoggingReloadOption
public override void Apply(NatsServer server)
{
server.Noticef("Reloaded: debug = {0}", _newValue);
// TODO: session 13 — call server.ReloadDebugRaftNodes(_newValue)
// DEFERRED: session 13 — call server.ReloadDebugRaftNodes(_newValue)
}
}
@@ -217,10 +222,10 @@ internal sealed class LogtimeReloadOption : LoggingReloadOption
/// Reload option for the <c>logtime_utc</c> setting.
/// Mirrors Go <c>logtimeUTCOption</c> struct in reload.go.
/// </summary>
internal sealed class LogtimeUtcReloadOption : LoggingReloadOption
internal sealed class LogtimeUTCOption : LoggingReloadOption
{
private readonly bool _newValue;
public LogtimeUtcReloadOption(bool newValue) => _newValue = newValue;
public LogtimeUTCOption(bool newValue) => _newValue = newValue;
public override void Apply(NatsServer server)
=> server.Noticef("Reloaded: logtime_utc = {0}", _newValue);
@@ -230,10 +235,10 @@ internal sealed class LogtimeUtcReloadOption : LoggingReloadOption
/// Reload option for the <c>log_file</c> setting.
/// Mirrors Go <c>logfileOption</c> struct in reload.go.
/// </summary>
internal sealed class LogFileReloadOption : LoggingReloadOption
internal sealed class LogfileOption : LoggingReloadOption
{
private readonly string _newValue;
public LogFileReloadOption(string newValue) => _newValue = newValue;
public LogfileOption(string newValue) => _newValue = newValue;
public override void Apply(NatsServer server)
=> server.Noticef("Reloaded: log_file = {0}", _newValue);
@@ -274,11 +279,11 @@ internal sealed class RemoteSyslogReloadOption : LoggingReloadOption
/// Mirrors Go <c>tlsOption</c> struct in reload.go.
/// The TLS config is stored as <c>object?</c> because the full
/// <c>TlsConfig</c> type is not yet ported.
/// TODO: session 13 — replace object? with the ported TlsConfig type.
/// DEFERRED: session 13 — replace object? with the ported TlsConfig type.
/// </summary>
internal sealed class TlsReloadOption : NoopReloadOption
{
// TODO: session 13 — replace object? with ported TlsConfig type
// DEFERRED: session 13 — replace object? with ported TlsConfig type
private readonly object? _newValue;
public TlsReloadOption(object? newValue) => _newValue = newValue;
@@ -288,7 +293,7 @@ internal sealed class TlsReloadOption : NoopReloadOption
{
var message = _newValue is null ? "disabled" : "enabled";
server.Noticef("Reloaded: tls = {0}", message);
// TODO: session 13 — update server.Info.TLSRequired / TLSVerify
// DEFERRED: session 13 — update server.Info.TLSRequired / TLSVerify
}
}
@@ -310,11 +315,11 @@ internal sealed class TlsTimeoutReloadOption : NoopReloadOption
/// Mirrors Go <c>tlsPinnedCertOption</c> struct in reload.go.
/// The pinned cert set is stored as <c>object?</c> pending the port
/// of the PinnedCertSet type.
/// TODO: session 13 — replace object? with ported PinnedCertSet type.
/// DEFERRED: session 13 — replace object? with ported PinnedCertSet type.
/// </summary>
internal sealed class TlsPinnedCertReloadOption : NoopReloadOption
{
// TODO: session 13 — replace object? with ported PinnedCertSet type
// DEFERRED: session 13 — replace object? with ported PinnedCertSet type
private readonly object? _newValue;
public TlsPinnedCertReloadOption(object? newValue) => _newValue = newValue;
@@ -459,17 +464,17 @@ internal sealed class AccountsReloadOption : AuthReloadOption
/// Reload option for the <c>cluster</c> setting.
/// Stores cluster options as <c>object?</c> pending the port of <c>ClusterOpts</c>.
/// Mirrors Go <c>clusterOption</c> struct in reload.go.
/// TODO: session 13 — replace object? with ported ClusterOpts type.
/// DEFERRED: session 13 — replace object? with ported ClusterOpts type.
/// </summary>
internal sealed class ClusterReloadOption : AuthReloadOption
{
// TODO: session 13 — replace object? with ported ClusterOpts type
// DEFERRED: session 13 — replace object? with ported ClusterOpts type
private readonly object? _newValue;
private readonly bool _permsChanged;
private readonly bool _poolSizeChanged;
private bool _poolSizeChanged;
private readonly bool _compressChanged;
private readonly string[] _accsAdded;
private readonly string[] _accsRemoved;
private string[] _accsAdded;
private string[] _accsRemoved;
public ClusterReloadOption(
object? newValue,
@@ -493,9 +498,34 @@ internal sealed class ClusterReloadOption : AuthReloadOption
public override bool IsClusterPoolSizeOrAccountsChange()
=> _poolSizeChanged || _accsAdded.Length > 0 || _accsRemoved.Length > 0;
/// <summary>
/// Computes pool/account deltas used by reload orchestration.
/// Mirrors Go <c>clusterOption.diffPoolAndAccounts</c>.
/// </summary>
public void DiffPoolAndAccounts(ClusterOpts oldValue)
{
ArgumentNullException.ThrowIfNull(oldValue);
if (_newValue is not ClusterOpts newValue)
{
_poolSizeChanged = false;
_accsAdded = [];
_accsRemoved = [];
return;
}
_poolSizeChanged = newValue.PoolSize != oldValue.PoolSize;
var oldAccounts = new HashSet<string>(oldValue.PinnedAccounts, StringComparer.Ordinal);
var newAccounts = new HashSet<string>(newValue.PinnedAccounts, StringComparer.Ordinal);
_accsAdded = newAccounts.Except(oldAccounts).ToArray();
_accsRemoved = oldAccounts.Except(newAccounts).ToArray();
}
public override void Apply(NatsServer server)
{
// TODO: session 13 — full cluster apply logic (TLS, route info, compression)
// DEFERRED: session 13 — full cluster apply logic (TLS, route info, compression)
server.Noticef("Reloaded: cluster");
}
}
@@ -504,11 +534,11 @@ internal sealed class ClusterReloadOption : AuthReloadOption
/// Reload option for the cluster <c>routes</c> setting.
/// Routes to add/remove are stored as <c>object[]</c> pending the port of URL handling.
/// Mirrors Go <c>routesOption</c> struct in reload.go.
/// TODO: session 13 — replace object[] with Uri[] when route types are ported.
/// DEFERRED: session 13 — replace object[] with Uri[] when route types are ported.
/// </summary>
internal sealed class RoutesReloadOption : NoopReloadOption
{
// TODO: session 13 — replace object[] with Uri[] when route URL types are ported
// DEFERRED: session 13 — replace object[] with Uri[] when route URL types are ported
private readonly object[] _add;
private readonly object[] _remove;
@@ -520,7 +550,7 @@ internal sealed class RoutesReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — add/remove routes, update varzUpdateRouteURLs
// DEFERRED: session 13 — add/remove routes, update varzUpdateRouteURLs
server.Noticef("Reloaded: cluster routes");
}
}
@@ -540,7 +570,7 @@ internal sealed class MaxConnReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — close random connections if over limit
// DEFERRED: session 13 — close random connections if over limit
server.Noticef("Reloaded: max_connections = {0}", _newValue);
}
}
@@ -558,7 +588,7 @@ internal sealed class PidFileReloadOption : NoopReloadOption
{
if (string.IsNullOrEmpty(_newValue))
return;
// TODO: session 13 — call server.LogPid()
// DEFERRED: session 13 — call server.LogPid()
server.Noticef("Reloaded: pid_file = {0}", _newValue);
}
}
@@ -580,7 +610,7 @@ internal sealed class PortsFileDirReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — call server.DeletePortsFile(_oldValue) and server.LogPorts()
// DEFERRED: session 13 — call server.DeletePortsFile(_oldValue) and server.LogPorts()
server.Noticef("Reloaded: ports_file_dir = {0}", _newValue);
}
}
@@ -596,7 +626,7 @@ internal sealed class MaxControlLineReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — update mcl on each connected client
// DEFERRED: session 13 — update mcl on each connected client
server.Noticef("Reloaded: max_control_line = {0}", _newValue);
}
}
@@ -612,7 +642,7 @@ internal sealed class MaxPayloadReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — update server info and mpay on each client
// DEFERRED: session 13 — update server info and mpay on each client
server.Noticef("Reloaded: max_payload = {0}", _newValue);
}
}
@@ -667,7 +697,7 @@ internal sealed class ClientAdvertiseReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — call server.SetInfoHostPort()
// DEFERRED: session 13 — call server.SetInfoHostPort()
server.Noticef("Reload: client_advertise = {0}", _newValue);
}
}
@@ -713,11 +743,11 @@ internal sealed class DefaultSentinelReloadOption : NoopReloadOption
/// Reload option for the OCSP setting.
/// The new value is stored as <c>object?</c> pending the port of <c>OCSPConfig</c>.
/// Mirrors Go <c>ocspOption</c> struct in reload.go.
/// TODO: session 13 — replace object? with ported OcspConfig type.
/// DEFERRED: session 13 — replace object? with ported OcspConfig type.
/// </summary>
internal sealed class OcspReloadOption : TlsBaseReloadOption
{
// TODO: session 13 — replace object? with ported OcspConfig type
// DEFERRED: session 13 — replace object? with ported OcspConfig type
private readonly object? _newValue;
public OcspReloadOption(object? newValue) => _newValue = newValue;
@@ -730,11 +760,11 @@ internal sealed class OcspReloadOption : TlsBaseReloadOption
/// The new value is stored as <c>object?</c> pending the port of
/// <c>OCSPResponseCacheConfig</c>.
/// Mirrors Go <c>ocspResponseCacheOption</c> struct in reload.go.
/// TODO: session 13 — replace object? with ported OcspResponseCacheConfig type.
/// DEFERRED: session 13 — replace object? with ported OcspResponseCacheConfig type.
/// </summary>
internal sealed class OcspResponseCacheReloadOption : TlsBaseReloadOption
{
// TODO: session 13 — replace object? with ported OcspResponseCacheConfig type
// DEFERRED: session 13 — replace object? with ported OcspResponseCacheConfig type
private readonly object? _newValue;
public OcspResponseCacheReloadOption(object? newValue) => _newValue = newValue;
@@ -779,7 +809,7 @@ internal sealed class MaxTracedMsgLenReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — update server.Opts.MaxTracedMsgLen under lock
// DEFERRED: session 13 — update server.Opts.MaxTracedMsgLen under lock
server.Noticef("Reloaded: max_traced_msg_len = {0}", _newValue);
}
}
@@ -812,7 +842,7 @@ internal sealed class MqttMaxAckPendingReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — call server.MqttUpdateMaxAckPending(_newValue)
// DEFERRED: session 13 — call server.MqttUpdateMaxAckPending(_newValue)
server.Noticef("Reloaded: MQTT max_ack_pending = {0}", _newValue);
}
}
@@ -884,7 +914,7 @@ internal sealed class ProfBlockRateReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — call server.SetBlockProfileRate(_newValue)
// DEFERRED: session 13 — call server.SetBlockProfileRate(_newValue)
server.Noticef("Reloaded: prof_block_rate = {0}", _newValue);
}
}
@@ -912,7 +942,7 @@ internal sealed class LeafNodeReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — full leaf-node apply logic from Go leafNodeOption.Apply()
// DEFERRED: session 13 — full leaf-node apply logic from Go leafNodeOption.Apply()
if (_tlsFirstChanged)
server.Noticef("Reloaded: LeafNode TLS HandshakeFirst settings");
if (_compressionChanged)
@@ -963,7 +993,7 @@ internal sealed class ProxiesReloadOption : NoopReloadOption
public override void Apply(NatsServer server)
{
// TODO: session 13 — disconnect proxied clients for removed keys,
// DEFERRED: session 13 — disconnect proxied clients for removed keys,
// call server.ProcessProxiesTrustedKeys()
if (_del.Length > 0)
server.Noticef("Reloaded: proxies trusted keys {0} were removed", string.Join(", ", _del));
@@ -985,7 +1015,7 @@ internal sealed class ProxiesReloadOption : NoopReloadOption
/// </summary>
internal sealed class ConfigReloader
{
// TODO: session 13 — full reload logic
// DEFERRED: session 13 — full reload logic
// Mirrors Go server.Reload() / server.ReloadOptions() in server/reload.go
/// <summary>
@@ -993,4 +1023,300 @@ internal sealed class ConfigReloader
/// Returns null on success; a non-null Exception describes the failure.
/// </summary>
public Exception? Reload(NatsServer server) => null;
/// <summary>
/// Applies bool-valued config precedence for reload:
/// config-file explicit values first, then explicit command-line flags.
/// Mirrors Go <c>applyBoolFlags</c>.
/// </summary>
public static void ApplyBoolFlags(ServerOptions newOptions, ServerOptions flagOptions)
{
ArgumentNullException.ThrowIfNull(newOptions);
ArgumentNullException.ThrowIfNull(flagOptions);
foreach (var (name, value) in newOptions.InConfig)
SetBooleanMember(newOptions, name, value);
foreach (var (name, value) in flagOptions.InCmdLine)
SetBooleanMember(newOptions, name, value);
}
/// <summary>
/// Sorts order-insensitive values in place prior to deep comparisons.
/// Mirrors Go <c>imposeOrder</c>.
/// </summary>
public static Exception? ImposeOrder(object? value)
{
switch (value)
{
case List<Account> accounts:
accounts.Sort((a, b) => string.CompareOrdinal(a.Name, b.Name));
break;
case List<User> users:
users.Sort((a, b) => string.CompareOrdinal(a.Username, b.Username));
break;
case List<NkeyUser> nkeys:
nkeys.Sort((a, b) => string.CompareOrdinal(a.Nkey, b.Nkey));
break;
case List<Uri> urls:
urls.Sort((a, b) => string.CompareOrdinal(a.ToString(), b.ToString()));
break;
case List<string> strings:
strings.Sort(StringComparer.Ordinal);
break;
case GatewayOpts gateway:
gateway.Gateways.Sort((a, b) => string.CompareOrdinal(a.Name, b.Name));
break;
case WebsocketOpts websocket:
websocket.AllowedOrigins.Sort(StringComparer.Ordinal);
break;
case null:
case string:
case bool:
case byte:
case ushort:
case uint:
case ulong:
case int:
case long:
case TimeSpan:
case float:
case double:
case LeafNodeOpts:
case ClusterOpts:
case SslServerAuthenticationOptions:
case PinnedCertSet:
case IAccountResolver:
case MqttOpts:
case Dictionary<string, string>:
case JsLimitOpts:
case StoreCipher:
case OcspResponseCacheConfig:
case ProxiesConfig:
case WriteTimeoutPolicy:
case AuthCalloutOpts:
case JsTpmOpts:
break;
default:
return new InvalidOperationException(
$"OnReload, sort or explicitly skip type: {value.GetType().FullName}");
}
return null;
}
/// <summary>
/// Returns remote gateway configs copied for diffing, with TLS runtime fields stripped.
/// Mirrors Go <c>copyRemoteGWConfigsWithoutTLSConfig</c>.
/// </summary>
public static List<RemoteGatewayOpts>? CopyRemoteGWConfigsWithoutTLSConfig(List<RemoteGatewayOpts>? current)
{
if (current is not { Count: > 0 })
return null;
var copied = new List<RemoteGatewayOpts>(current.Count);
foreach (var config in current)
{
copied.Add(new RemoteGatewayOpts
{
Name = config.Name,
TlsConfig = null,
TlsConfigOpts = null,
TlsTimeout = config.TlsTimeout,
Urls = [.. config.Urls.Select(static u => new Uri(u.ToString(), UriKind.Absolute))],
});
}
return copied;
}
/// <summary>
/// Returns remote leaf-node configs copied for diffing, with runtime-mutated
/// fields stripped or normalized.
/// Mirrors Go <c>copyRemoteLNConfigForReloadCompare</c>.
/// </summary>
public static List<RemoteLeafOpts>? CopyRemoteLNConfigForReloadCompare(List<RemoteLeafOpts>? current)
{
if (current is not { Count: > 0 })
return null;
var copied = new List<RemoteLeafOpts>(current.Count);
foreach (var config in current)
{
copied.Add(new RemoteLeafOpts
{
LocalAccount = config.LocalAccount,
NoRandomize = config.NoRandomize,
Urls = [.. config.Urls.Select(static u => new Uri(u.ToString(), UriKind.Absolute))],
Credentials = config.Credentials,
Nkey = config.Nkey,
SignatureCb = config.SignatureCb,
Tls = false,
TlsConfig = null,
TlsConfigOpts = null,
TlsTimeout = config.TlsTimeout,
TlsHandshakeFirst = false,
Hub = config.Hub,
DenyImports = [],
DenyExports = [],
FirstInfoTimeout = config.FirstInfoTimeout,
Compression = new CompressionOpts(),
Websocket = new RemoteLeafWebsocketOpts
{
Compression = config.Websocket.Compression,
NoMasking = config.Websocket.NoMasking,
},
Proxy = new RemoteLeafProxyOpts
{
Url = config.Proxy.Url,
Username = config.Proxy.Username,
Password = config.Proxy.Password,
Timeout = config.Proxy.Timeout,
},
JetStreamClusterMigrate = config.JetStreamClusterMigrate,
JetStreamClusterMigrateDelay = config.JetStreamClusterMigrateDelay,
LocalIsolation = config.LocalIsolation,
RequestIsolation = config.RequestIsolation,
Disabled = false,
});
}
return copied;
}
/// <summary>
/// Validates non-reloadable cluster settings and advertise syntax.
/// Mirrors Go <c>validateClusterOpts</c>.
/// </summary>
public static Exception? ValidateClusterOpts(ClusterOpts oldValue, ClusterOpts newValue)
{
ArgumentNullException.ThrowIfNull(oldValue);
ArgumentNullException.ThrowIfNull(newValue);
if (!string.Equals(oldValue.Host, newValue.Host, StringComparison.Ordinal))
{
return new InvalidOperationException(
$"config reload not supported for cluster host: old={oldValue.Host}, new={newValue.Host}");
}
if (oldValue.Port != newValue.Port)
{
return new InvalidOperationException(
$"config reload not supported for cluster port: old={oldValue.Port}, new={newValue.Port}");
}
if (!string.IsNullOrEmpty(newValue.Advertise))
{
var (_, _, err) = ServerUtilities.ParseHostPort(newValue.Advertise, 0);
if (err != null)
return new InvalidOperationException(
$"invalid Cluster.Advertise value of {newValue.Advertise}, err={err.Message}", err);
}
return null;
}
/// <summary>
/// Diffs old/new route lists and returns routes to add/remove.
/// Mirrors Go <c>diffRoutes</c>.
/// </summary>
public static (List<Uri> Add, List<Uri> Remove) DiffRoutes(IReadOnlyList<Uri> oldRoutes, IReadOnlyList<Uri> newRoutes)
{
ArgumentNullException.ThrowIfNull(oldRoutes);
ArgumentNullException.ThrowIfNull(newRoutes);
var add = new List<Uri>();
var remove = new List<Uri>();
foreach (var oldRoute in oldRoutes)
{
if (!newRoutes.Any(newRoute => ServerUtilities.UrlsAreEqual(oldRoute, newRoute)))
remove.Add(oldRoute);
}
foreach (var newRoute in newRoutes)
{
if (!oldRoutes.Any(oldRoute => ServerUtilities.UrlsAreEqual(oldRoute, newRoute)))
add.Add(newRoute);
}
return (add, remove);
}
/// <summary>
/// Diffs proxy trusted keys and returns added/removed key sets.
/// Mirrors Go <c>diffProxiesTrustedKeys</c>.
/// </summary>
public static (List<string> Add, List<string> Del) DiffProxiesTrustedKeys(
IReadOnlyList<ProxyConfig>? oldTrusted,
IReadOnlyList<ProxyConfig>? newTrusted)
{
var oldList = oldTrusted ?? [];
var newList = newTrusted ?? [];
var add = new List<string>();
var del = new List<string>();
foreach (var oldProxy in oldList)
{
if (!newList.Any(np => string.Equals(np.Key, oldProxy.Key, StringComparison.Ordinal)))
del.Add(oldProxy.Key);
}
foreach (var newProxy in newList)
{
if (!oldList.Any(op => string.Equals(op.Key, newProxy.Key, StringComparison.Ordinal)))
add.Add(newProxy.Key);
}
return (add, del);
}
private static void SetBooleanMember(ServerOptions options, string path, bool value)
{
var segments = path.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
if (segments.Length == 0)
return;
object? target = options;
for (int i = 0; i < segments.Length - 1; i++)
{
if (target == null)
return;
var segment = segments[i];
var targetType = target.GetType();
var prop = targetType.GetProperty(segment, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic);
if (prop != null)
{
target = prop.GetValue(target);
continue;
}
var field = targetType.GetField(segment, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic);
if (field != null)
{
target = field.GetValue(target);
continue;
}
return;
}
if (target == null)
return;
var leaf = segments[^1];
var leafType = target.GetType();
var leafProperty = leafType.GetProperty(leaf, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic);
if (leafProperty?.PropertyType == typeof(bool) && leafProperty.CanWrite)
{
leafProperty.SetValue(target, value);
return;
}
var leafField = leafType.GetField(leaf, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic);
if (leafField?.FieldType == typeof(bool))
leafField.SetValue(target, value);
}
}