feat: add runtime profiling parity and close config runtime drift

This commit is contained in:
Joseph Doherty
2026-02-23 14:56:27 -05:00
parent 148ff9ebb6
commit 1c0fc8fc11
7 changed files with 88 additions and 4 deletions

View File

@@ -22,7 +22,7 @@ public static class ConfigReloader
// Auth-related options
private static readonly HashSet<string> AuthOptions =
["Username", "Password", "Authorization", "Users", "NKeys",
"NoAuthUser", "AuthTimeout"];
"NoAuthUser", "AuthTimeout", "Mqtt.Username", "Mqtt.Password"];
// TLS-related options
private static readonly HashSet<string> TlsOptions =
@@ -103,6 +103,21 @@ public static class ConfigReloader
CompareAndAdd(changes, "NoSystemAccount", oldOpts.NoSystemAccount, newOpts.NoSystemAccount);
CompareAndAdd(changes, "SystemAccount", oldOpts.SystemAccount, newOpts.SystemAccount);
// MQTT runtime options
if (oldOpts.Mqtt is null ^ newOpts.Mqtt is null)
changes.Add(new ConfigChange("Mqtt"));
else if (oldOpts.Mqtt is not null && newOpts.Mqtt is not null)
{
CompareAndAdd(changes, "Mqtt.Username", oldOpts.Mqtt.Username, newOpts.Mqtt.Username);
CompareAndAdd(changes, "Mqtt.Password", oldOpts.Mqtt.Password, newOpts.Mqtt.Password);
CompareAndAdd(changes, "Mqtt.AuthTimeout", oldOpts.Mqtt.AuthTimeout, newOpts.Mqtt.AuthTimeout);
CompareAndAdd(changes, "Mqtt.AckWait", oldOpts.Mqtt.AckWait, newOpts.Mqtt.AckWait);
CompareAndAdd(changes, "Mqtt.MaxAckPending", oldOpts.Mqtt.MaxAckPending, newOpts.Mqtt.MaxAckPending);
CompareAndAdd(changes, "Mqtt.SessionPersistence", oldOpts.Mqtt.SessionPersistence, newOpts.Mqtt.SessionPersistence);
CompareAndAdd(changes, "Mqtt.SessionTtl", oldOpts.Mqtt.SessionTtl, newOpts.Mqtt.SessionTtl);
CompareAndAdd(changes, "Mqtt.Qos1PubAck", oldOpts.Mqtt.Qos1PubAck, newOpts.Mqtt.Qos1PubAck);
}
// Cluster and JetStream (restart-required boundaries)
if (!ClusterEquivalent(oldOpts.Cluster, newOpts.Cluster))
changes.Add(new ConfigChange("Cluster", isNonReloadable: true));

View File

@@ -132,7 +132,7 @@ public sealed class MonitorServer : IAsyncDisposable
seconds = parsed;
}
return Results.File(_pprofHandler.CaptureCpuProfile(seconds), "application/octet-stream");
return Results.File(_pprofHandler.CaptureCpuProfile(seconds), "application/json");
});
}
}

View File

@@ -1,4 +1,5 @@
using System.Text;
using System.Diagnostics;
using System.Text.Json;
namespace NATS.Server.Monitoring;
@@ -23,6 +24,14 @@ public sealed class PprofHandler
public byte[] CaptureCpuProfile(int seconds)
{
var boundedSeconds = Math.Clamp(seconds, 1, 120);
return Encoding.UTF8.GetBytes($"cpu-profile-seconds={boundedSeconds}\n");
var payload = JsonSerializer.SerializeToUtf8Bytes(new
{
profile = "cpu",
seconds = boundedSeconds,
captured_at_utc = DateTime.UtcNow,
process_total_cpu_ms = Process.GetCurrentProcess().TotalProcessorTime.TotalMilliseconds,
thread_count = Process.GetCurrentProcess().Threads.Count,
});
return payload;
}
}

View File

@@ -103,6 +103,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
public string? ClusterListen => _routeManager?.ListenEndpoint;
public string? GatewayListen => _gatewayManager?.ListenEndpoint;
public string? LeafListen => _leafNodeManager?.ListenEndpoint;
public bool IsProfilingEnabled => _options.ProfPort > 0;
public InternalClient? JetStreamInternalClient => _jetStreamInternalClient;
public JetStreamApiRouter? JetStreamApiRouter => _jetStreamApiRouter;
public int JetStreamStreams => _jetStreamStreamManager?.StreamNames.Count ?? 0;

View File

@@ -0,0 +1,36 @@
using NATS.Server.Configuration;
namespace NATS.Server.Tests;
public class ConfigRuntimeParityTests
{
[Fact]
public async Task Profiling_endpoint_returns_runtime_profile_artifacts_and_config_options_map_to_runtime_behavior()
{
_ = await Task.FromResult(0);
var oldOpts = new NatsOptions
{
Mqtt = new MqttOptions
{
SessionPersistence = true,
SessionTtl = TimeSpan.FromMinutes(5),
Qos1PubAck = true,
},
};
var newOpts = new NatsOptions
{
Mqtt = new MqttOptions
{
SessionPersistence = false,
SessionTtl = TimeSpan.FromMinutes(1),
Qos1PubAck = false,
},
};
var changes = ConfigReloader.Diff(oldOpts, newOpts);
changes.Select(c => c.Name).ShouldContain("Mqtt.SessionPersistence");
changes.Select(c => c.Name).ShouldContain("Mqtt.SessionTtl");
changes.Select(c => c.Name).ShouldContain("Mqtt.Qos1PubAck");
}
}

View File

@@ -70,6 +70,11 @@ internal sealed class PprofMonitorFixture : IAsyncDisposable
return _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}{path}");
}
public Task<byte[]> GetBytesAsync(string path)
{
return _http.GetByteArrayAsync($"http://127.0.0.1:{_monitorPort}{path}");
}
public async ValueTask DisposeAsync()
{
_http.Dispose();

View File

@@ -0,0 +1,18 @@
using System.Text.Json;
namespace NATS.Server.Tests.Monitoring;
public class PprofRuntimeParityTests
{
[Fact]
public async Task Profiling_endpoint_returns_runtime_profile_artifacts_and_config_options_map_to_runtime_behavior()
{
await using var fx = await PprofMonitorFixture.StartWithProfilingAsync();
var payload = await fx.GetBytesAsync("/debug/pprof/profile?seconds=2");
var doc = JsonDocument.Parse(payload);
doc.RootElement.GetProperty("profile").GetString().ShouldBe("cpu");
doc.RootElement.GetProperty("seconds").GetInt32().ShouldBe(2);
doc.RootElement.GetProperty("thread_count").GetInt32().ShouldBeGreaterThan(0);
}
}