natsdotnet/tests/NATS.Server.Tests/Monitoring/ConnzParityTests.cs

// Ported from golang/nats-server/server/monitor_test.go
// TestMonitorConnz — verify /connz lists active connections with correct fields.
// TestMonitorConnzSortedByBytesAndMsgs — verify /connz?sort=bytes_to ordering.

using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;

namespace NATS.Server.Tests;

public class ConnzParityTests : IAsyncLifetime
{
    private readonly NatsServer _server;
    private readonly int _natsPort;
    private readonly int _monitorPort;
    private readonly CancellationTokenSource _cts = new();
    private readonly HttpClient _http = new();

    public ConnzParityTests()
    {
        _natsPort = GetFreePort();
        _monitorPort = GetFreePort();
        _server = new NatsServer(
            new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
            NullLoggerFactory.Instance);
    }

    public async Task InitializeAsync()
    {
        _ = _server.StartAsync(_cts.Token);
        await _server.WaitForReadyAsync();
        for (var i = 0; i < 50; i++)
        {
            try
            {
                var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
                if (probe.IsSuccessStatusCode) break;
            }
            catch (HttpRequestException) { }
            await Task.Delay(50);
        }
    }

    public async Task DisposeAsync()
    {
        _http.Dispose();
        await _cts.CancelAsync();
        _server.Dispose();
    }

    /// <summary>
    /// Corresponds to Go TestMonitorConnz.
    /// Verifies /connz lists active connections and that per-connection fields
    /// (ip, port, lang, version, uptime) are populated once 2 clients are connected.
    /// </summary>
    [Fact]
    public async Task Connz_lists_active_connections()
    {
        var sockets = new List<Socket>();
        try
        {
            // Connect 2 named clients
            for (var i = 0; i < 2; i++)
            {
                var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
                var ns = new NetworkStream(sock);
                var buf = new byte[4096];
                _ = await ns.ReadAsync(buf); // consume INFO
                var connect = $"CONNECT {{\"name\":\"client-{i}\",\"lang\":\"csharp\",\"version\":\"1.0\"}}\r\n";
                await ns.WriteAsync(System.Text.Encoding.ASCII.GetBytes(connect));
                await ns.FlushAsync();
                sockets.Add(sock);
            }

            await Task.Delay(200);

            var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz");
            response.StatusCode.ShouldBe(HttpStatusCode.OK);

            var connz = await response.Content.ReadFromJsonAsync<Connz>();
            connz.ShouldNotBeNull();

            // Both clients must appear
            connz.NumConns.ShouldBeGreaterThanOrEqualTo(2);
            connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);

            // Verify per-connection identity fields on one of our named connections
            var conn = connz.Conns.First(c => c.Name == "client-0");
            conn.Ip.ShouldNotBeNullOrEmpty();
            conn.Port.ShouldBeGreaterThan(0);
            conn.Lang.ShouldBe("csharp");
            conn.Version.ShouldBe("1.0");
            conn.Uptime.ShouldNotBeNullOrEmpty();
        }
        finally
        {
            foreach (var s in sockets) s.Dispose();
        }
    }

    /// <summary>
    /// Corresponds to Go TestMonitorConnzSortedByBytesAndMsgs (bytes_to / out_bytes ordering).
    /// Connects a high-traffic client that publishes 100 messages and 3 baseline clients,
    /// then verifies /connz?sort=bytes_to returns connections in descending out_bytes order.
    /// </summary>
    [Fact]
    public async Task Connz_sort_by_bytes()
    {
        var sockets = new List<(Socket Sock, NetworkStream Ns)>();
        try
        {
            // Connect a subscriber first so that published messages are delivered (and counted as out_bytes)
            var subSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            await subSock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
            var subNs = new NetworkStream(subSock);
            var subBuf = new byte[4096];
            _ = await subNs.ReadAsync(subBuf);
            await subNs.WriteAsync("CONNECT {}\r\nSUB foo 1\r\n"u8.ToArray());
            await subNs.FlushAsync();
            sockets.Add((subSock, subNs));

            // High-traffic publisher: publish 100 messages to "foo"
            var highSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            await highSock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
            var highNs = new NetworkStream(highSock);
            var highBuf = new byte[4096];
            _ = await highNs.ReadAsync(highBuf);
            await highNs.WriteAsync("CONNECT {}\r\n"u8.ToArray());
            await highNs.FlushAsync();

            for (var i = 0; i < 100; i++)
                await highNs.WriteAsync("PUB foo 11\r\nHello World\r\n"u8.ToArray());
            await highNs.FlushAsync();
            sockets.Add((highSock, highNs));

            // 3 baseline clients — no traffic beyond CONNECT
            for (var i = 0; i < 3; i++)
            {
                var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
                var ns = new NetworkStream(sock);
                var buf = new byte[4096];
                _ = await ns.ReadAsync(buf);
                await ns.WriteAsync("CONNECT {}\r\n"u8.ToArray());
                await ns.FlushAsync();
                sockets.Add((sock, ns));
            }

            await Task.Delay(300);

            var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=bytes_to");
            response.StatusCode.ShouldBe(HttpStatusCode.OK);

            var connz = await response.Content.ReadFromJsonAsync<Connz>();
            connz.ShouldNotBeNull();
            connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);

            // The first entry must have at least as many out_bytes as the second (descending order)
            connz.Conns[0].OutBytes.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].OutBytes);
        }
        finally
        {
            foreach (var (s, _) in sockets) s.Dispose();
        }
    }

    private static int GetFreePort()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
        return ((IPEndPoint)sock.LocalEndPoint!).Port;
    }
}