diff --git a/proto.lock b/proto.lock index 5ea263371e..6e1017d5e8 100644 --- a/proto.lock +++ b/proto.lock @@ -1554,21 +1554,21 @@ ] }, { - "name": "TcpStatsReq" + "name": "ConnectionStatsReq" }, { - "name": "TcpStatsResp", + "name": "ConnectionStatsResp", "fields": [ { "id": 1, "name": "connections", - "type": "TcpConnectionStats", + "type": "ConnectionStats", "is_repeated": true } ] }, { - "name": "TcpConnectionStats", + "name": "ConnectionStats", "fields": [ { "id": 1, @@ -1603,22 +1603,32 @@ { "id": 7, "name": "pending_send_bytes", - "type": "int32" + "type": "int64" }, { "id": 8, "name": "pending_received_bytes", - "type": "int32" + "type": "int64" }, { "id": 9, - "name": "is_external_connection", + "name": "is_tls", "type": "bool" }, { "id": 10, - "name": "is_ssl_connection", - "type": "bool" + "name": "protocol", + "type": "string" + }, + { + "id": 11, + "name": "application", + "type": "string" + }, + { + "id": 12, + "name": "connected_at", + "type": "google.protobuf.Timestamp" } ] }, @@ -1693,9 +1703,9 @@ "out_streamed": true }, { - "name": "TcpStats", - "in_type": "TcpStatsReq", - "out_type": "TcpStatsResp" + "name": "ConnectionStats", + "in_type": "ConnectionStatsReq", + "out_type": "ConnectionStatsResp" }, { "name": "ReplicationStats", @@ -1708,6 +1718,9 @@ "imports": [ { "path": "google/protobuf/struct.proto" + }, + { + "path": "google/protobuf/timestamp.proto" } ], "package": { diff --git a/src/EventStore.ClusterNode/Components/Pages/Observability.razor b/src/EventStore.ClusterNode/Components/Pages/Observability.razor index 8d3540afe8..f244b0379f 100644 --- a/src/EventStore.ClusterNode/Components/Pages/Observability.razor +++ b/src/EventStore.ClusterNode/Components/Pages/Observability.razor @@ -10,7 +10,7 @@

Observability

A map of the signals the node already emits.

-

Inspect live queue pressure, processing throughput, health, metrics, replication, TCP, and grouped runtime statistics.

+

Inspect live queue pressure, processing throughput, health, metrics, replication, and grouped runtime statistics.

@@ -20,6 +20,116 @@
+
+
+
+

Network boundary

+

Shared HTTP and gRPC connections

+

Active connections accepted by the node endpoint, including per-second traffic rates and pending bytes.

+
+
+ @NetworkStatusLabel + @NetworkPageStatusLabel + + +
+
+ +
+
+ + + + + + + + + + + + + + + + + @if (NetworkRows.Count == 0) + { + + } + else + { + foreach (var connection in NetworkRows.Take(5)) + { + + + + + + + + + + + + + } + } + +
ConnectionClientTypeRemote endpointSent rateSent currentSent pendingReceived rateReceived currentReceived pending
@NetworkEmptyMessage
@connection.ConnectionId@Display(connection.ClientName)@connection.Application · @Display(connection.Protocol) · @(connection.IsTls ? "TLS" : "Cleartext")@connection.RemoteEndPointWaiting@FormatBytes(connection.TotalBytesSent)@FormatBytes(connection.PendingSendBytes)Waiting@FormatBytes(connection.TotalBytesReceived)@FormatBytes(connection.PendingReceivedBytes)
+
+
+
+ +
+
+

Cluster transport

+

gRPC replication connections

+

Live database replication sessions on the shared HTTP/2 endpoint.

+
+ +
+
+ + + + + + + + + + + + + @if (Page is not null && !string.IsNullOrWhiteSpace(Page.ReplicationMessage)) + { + + } + else if (Page is not null && Page.ReplicationConnections.Count > 0) + { + @foreach (var connection in Page.ReplicationConnections) + { + + + + + + + + + } + } + else + { + + } + +
EndpointConnectionSentReceivedPendingSend queue
@Page.ReplicationMessage
@connection.Endpoint@connection.ConnectionId@FormatBytes(connection.TotalBytesSent)@FormatBytes(connection.TotalBytesReceived)@FormatBytes(connection.PendingSendBytes + connection.PendingReceivedBytes)@connection.SendQueueSize.ToString("N0", CultureInfo.InvariantCulture)
No active gRPC replication connections.
+
+
+
+
Dashboard snapshot @@ -87,58 +197,6 @@ }
-
-
-
-

TCP dashboard

-

Realtime connections

-

Shows active connections with per-second sent and received byte rates.

-
-
- @TcpStatusLabel - @TcpPageStatusLabel - - -
-
- -
-
- - - - - - - - - - - - - - - - - @if (TcpRows.Count == 0) - { - - - - } - else - { - foreach (var connection in TcpRows.Take(5)) - { - @RenderTcpRow(connection) - } - } - -
ConnectionClientTypeIP AddressSent rateSent currentSent pendingReceived rateReceived currentReceived pending
@TcpEmptyMessage
-
-
-
- @code { @@ -148,29 +206,30 @@ private string Expanded { get; set; } = ""; private string DashboardPayloadJson => Page?.ClientPayloadJson ?? "{}"; - private IReadOnlyList TcpRows => Page?.TcpConnections ?? Array.Empty(); - private string TcpErrorMessage => !string.IsNullOrWhiteSpace(Page?.TcpMessage) - ? Page.TcpMessage - : Page?.Message ?? ""; - private string TcpStatusLabel { - get { - if (Page is null) - return "Connecting TCP stats..."; - - if (!string.IsNullOrWhiteSpace(TcpErrorMessage)) - return "TCP unavailable"; - - return TcpRows.Count == 0 - ? "TCP live" - : string.Create(CultureInfo.InvariantCulture, $"TCP live · {TcpRows.Count} connection{(TcpRows.Count == 1 ? "" : "s")}"); - } - } - private string TcpPageStatusLabel => TcpRows.Count == 0 + private IReadOnlyList NetworkRows => + Page?.NodeConnections ?? Array.Empty(); + private string NetworkStatusLabel => Page is null + ? "Connecting network stats..." + : !Page.NetworkAvailable + ? "Network unavailable" + : NetworkRows.Count == 0 + ? "Network live" + : string.Create(CultureInfo.InvariantCulture, $"Network live · {NetworkRows.Count} connection{(NetworkRows.Count == 1 ? "" : "s")}"); + private string NetworkEmptyMessage => Page is null + ? "Waiting for network statistics..." + : Page.NetworkAvailable + ? "No active shared-endpoint connections." + : "Network statistics are unavailable."; + private string NetworkPageStatusLabel => NetworkRows.Count == 0 ? "No pages" - : string.Create(CultureInfo.InvariantCulture, $"Page 1 of {Math.Max(1, (int)Math.Ceiling(TcpRows.Count / 5.0))}"); - private string TcpEmptyMessage => string.IsNullOrWhiteSpace(TcpErrorMessage) - ? "No TCP connections are currently reported." - : TcpErrorMessage; + : string.Create(CultureInfo.InvariantCulture, $"Page 1 of {Math.Max(1, (int)Math.Ceiling(NetworkRows.Count / 5.0))}"); + + private static string FormatBytes(long value) => + value.ToString("N0", CultureInfo.InvariantCulture) + " B"; + private static string Display(string value) => string.IsNullOrWhiteSpace(value) ? "" : value; + private static string ConnectionDetails(NodeConnectionSnapshot connection) => + string.Create(CultureInfo.InvariantCulture, + $"Local {connection.LocalEndPoint}, connected {connection.ConnectedAt:u}"); protected override async Task OnParametersSetAsync() { Page = null; @@ -216,19 +275,6 @@ @row.CurrentLastMessageLabel ; - private RenderFragment RenderTcpRow(TcpConnectionRow connection) => @ - @connection.IdLabel - @connection.ClientLabel - @connection.TypeLabel - @connection.RemoteEndPointLabel - @connection.SentRateLabel - @connection.TotalBytesSentLabel - @connection.PendingSendBytesLabel - @connection.ReceivedRateLabel - @connection.TotalBytesReceivedLabel - @connection.PendingReceivedBytesLabel - ; - private bool IsExpanded(string groupName) => ParseExpandedGroups().Contains(groupName); diff --git a/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs b/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs new file mode 100644 index 0000000000..0a0fdfc005 --- /dev/null +++ b/src/EventStore.ClusterNode/Components/Services/NodeConnectionTracker.cs @@ -0,0 +1,303 @@ +using System; +using System.Buffers; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO.Pipelines; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Core.Services.Transport.Grpc; +using Microsoft.AspNetCore.Connections; + +namespace EventStore.ClusterNode.Components.Services; + +public sealed class NodeConnectionTracker : IConnectionStatsProvider +{ + private readonly ConcurrentDictionary _connections = new(); + + public IReadOnlyList Snapshot() => + _connections.Values.Select(x => x.Snapshot()) + .OrderBy(x => x.RemoteEndPoint, StringComparer.OrdinalIgnoreCase) + .ThenBy(x => x.ConnectionId, StringComparer.Ordinal) + .ToArray(); + + IReadOnlyList IConnectionStatsProvider.Snapshot() => Snapshot(); + + public async Task Track(ConnectionContext context, ConnectionDelegate next, bool isTls) + { + var state = new NodeConnectionState( + context.ConnectionId, + context.RemoteEndPoint?.ToString() ?? "", + context.LocalEndPoint?.ToString() ?? "", + isTls, + DateTimeOffset.UtcNow); + _connections[context.ConnectionId] = state; + context.Transport = new CountingDuplexPipe(context.Transport, state); + + try + { + await next(context); + } + finally + { + _connections.TryRemove(context.ConnectionId, out _); + } + } + + public void ObserveRequest( + string connectionId, + string protocol, + bool isGrpc, + string connectionName, + string userAgent) + { + if (_connections.TryGetValue(connectionId, out var connection)) + { + connection.ObserveRequest(protocol, isGrpc, connectionName, userAgent); + } + } +} + +public sealed record NodeConnectionSnapshot( + string ConnectionId, + string RemoteEndPoint, + string LocalEndPoint, + string ClientName, + string Application, + string Protocol, + bool IsTls, + DateTimeOffset ConnectedAt, + long TotalBytesSent, + long TotalBytesReceived, + long PendingSendBytes, + long PendingReceivedBytes) : ConnectionStatsSnapshot( + ConnectionId, + RemoteEndPoint, + LocalEndPoint, + ClientName, + Application, + Protocol, + IsTls, + ConnectedAt, + TotalBytesSent, + TotalBytesReceived, + PendingSendBytes, + PendingReceivedBytes); + +internal sealed class NodeConnectionState +{ + private readonly object _metadataLock = new(); + private string _clientName = ""; + private bool _hasExplicitConnectionName; + private bool _hasGrpcRequests; + private bool _hasHttpRequests; + private long _pendingReceivedBytes; + private long _pendingSendBytes; + private string _protocol = ""; + private long _totalBytesReceived; + private long _totalBytesSent; + + public NodeConnectionState( + string connectionId, + string remoteEndPoint, + string localEndPoint, + bool isTls, + DateTimeOffset connectedAt) + { + ConnectionId = connectionId; + RemoteEndPoint = remoteEndPoint; + LocalEndPoint = localEndPoint; + IsTls = isTls; + ConnectedAt = connectedAt; + } + + private string ConnectionId { get; } + private string RemoteEndPoint { get; } + private string LocalEndPoint { get; } + private bool IsTls { get; } + private DateTimeOffset ConnectedAt { get; } + + public void Received(long bytes, long pendingBytes) + { + Interlocked.Add(ref _totalBytesReceived, bytes); + Interlocked.Exchange(ref _pendingReceivedBytes, pendingBytes); + } + + public void Reading(long pendingBytes) => + Interlocked.Exchange(ref _pendingReceivedBytes, pendingBytes); + + public void Sending(int bytes) + { + Interlocked.Add(ref _totalBytesSent, bytes); + Interlocked.Add(ref _pendingSendBytes, bytes); + } + + public void Sent() => Interlocked.Exchange(ref _pendingSendBytes, 0); + + public void ObserveRequest( + string protocol, + bool isGrpc, + string connectionName, + string userAgent) + { + lock (_metadataLock) + { + _protocol = Merge(_protocol, protocol); + _hasGrpcRequests |= isGrpc; + _hasHttpRequests |= !isGrpc; + + if (!string.IsNullOrWhiteSpace(connectionName)) + { + _clientName = connectionName; + _hasExplicitConnectionName = true; + } + else if (!_hasExplicitConnectionName && !string.IsNullOrWhiteSpace(userAgent)) + { + _clientName = userAgent; + } + } + } + + public NodeConnectionSnapshot Snapshot() + { + lock (_metadataLock) + { + return new( + ConnectionId, + RemoteEndPoint, + LocalEndPoint, + _clientName, + ApplicationLabel(), + _protocol, + IsTls, + ConnectedAt, + Interlocked.Read(ref _totalBytesSent), + Interlocked.Read(ref _totalBytesReceived), + Interlocked.Read(ref _pendingSendBytes), + Interlocked.Read(ref _pendingReceivedBytes)); + } + } + + private string ApplicationLabel() => (_hasHttpRequests, _hasGrpcRequests) switch + { + (true, true) => "HTTP and gRPC", + (false, true) => "gRPC", + (true, false) => "HTTP", + _ => "Awaiting request" + }; + + private static string Merge(string current, string observed) + { + if (string.IsNullOrWhiteSpace(observed) || current == observed) + { + return current; + } + + return string.IsNullOrWhiteSpace(current) ? observed : "Mixed"; + } +} + +internal sealed class CountingDuplexPipe : IDuplexPipe +{ + public CountingDuplexPipe(IDuplexPipe inner, NodeConnectionState state) + { + Input = new CountingPipeReader(inner.Input, state); + Output = new CountingPipeWriter(inner.Output, state); + } + + public PipeReader Input { get; } + public PipeWriter Output { get; } +} + +internal sealed class CountingPipeReader : PipeReader +{ + private readonly PipeReader _inner; + private readonly NodeConnectionState _state; + private ReadOnlySequence _currentBuffer; + + public CountingPipeReader(PipeReader inner, NodeConnectionState state) + { + _inner = inner; + _state = state; + } + + public override void AdvanceTo(SequencePosition consumed) => AdvanceTo(consumed, consumed); + + public override void AdvanceTo(SequencePosition consumed, SequencePosition examined) + { + var consumedBytes = _currentBuffer.IsEmpty ? 0 : _currentBuffer.Slice(0, consumed).Length; + var pendingBytes = _currentBuffer.IsEmpty ? 0 : _currentBuffer.Slice(consumed).Length; + _state.Received(consumedBytes, pendingBytes); + _currentBuffer = default; + _inner.AdvanceTo(consumed, examined); + } + + public override void CancelPendingRead() => _inner.CancelPendingRead(); + + public override void Complete(Exception exception = null) => _inner.Complete(exception); + + public override ValueTask CompleteAsync(Exception exception = null) => _inner.CompleteAsync(exception); + + public override async ValueTask ReadAsync(CancellationToken cancellationToken = default) + { + var result = await _inner.ReadAsync(cancellationToken); + Observe(result); + return result; + } + + public override bool TryRead(out ReadResult result) + { + if (!_inner.TryRead(out result)) + { + return false; + } + + Observe(result); + return true; + } + + private void Observe(ReadResult result) + { + _currentBuffer = result.Buffer; + _state.Reading(result.Buffer.Length); + } +} + +internal sealed class CountingPipeWriter : PipeWriter +{ + private readonly PipeWriter _inner; + private readonly NodeConnectionState _state; + + public CountingPipeWriter(PipeWriter inner, NodeConnectionState state) + { + _inner = inner; + _state = state; + } + + public override void Advance(int bytes) + { + _state.Sending(bytes); + _inner.Advance(bytes); + } + + public override void CancelPendingFlush() => _inner.CancelPendingFlush(); + + public override void Complete(Exception exception = null) => _inner.Complete(exception); + + public override ValueTask CompleteAsync(Exception exception = null) => _inner.CompleteAsync(exception); + + public override async ValueTask FlushAsync(CancellationToken cancellationToken = default) + { + var result = await _inner.FlushAsync(cancellationToken); + if (!result.IsCanceled) + { + _state.Sent(); + } + + return result; + } + + public override Memory GetMemory(int sizeHint = 0) => _inner.GetMemory(sizeHint); + + public override Span GetSpan(int sizeHint = 0) => _inner.GetSpan(sizeHint); +} diff --git a/src/EventStore.ClusterNode/Components/Services/QueueDashboardService.cs b/src/EventStore.ClusterNode/Components/Services/QueueDashboardService.cs index 6cba5ae36b..1383231321 100644 --- a/src/EventStore.ClusterNode/Components/Services/QueueDashboardService.cs +++ b/src/EventStore.ClusterNode/Components/Services/QueueDashboardService.cs @@ -18,23 +18,23 @@ public sealed class QueueDashboardService { private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(10); private static readonly Operation StatisticsOperation = new(Operations.Node.Statistics.Read); - private static readonly Operation TcpStatisticsOperation = new(Operations.Node.Statistics.Tcp); + private static readonly Operation ReplicationStatisticsOperation = new(Operations.Node.Statistics.Replication); private readonly IAuthorizationProvider _authorizationProvider; private readonly IHttpContextAccessor _httpContextAccessor; private readonly IPublisher _monitoringQueue; - private readonly object _tcpGate = new(); - private Dictionary _previousTcpConnections = new(); - private DateTime? _lastTcpRefresh; + private readonly NodeConnectionTracker _nodeConnectionTracker; public QueueDashboardService( IAuthorizationProvider authorizationProvider, IHttpContextAccessor httpContextAccessor, - StandardComponents standardComponents) + StandardComponents standardComponents, + NodeConnectionTracker nodeConnectionTracker) { _authorizationProvider = authorizationProvider; _httpContextAccessor = httpContextAccessor; _monitoringQueue = standardComponents.MonitoringQueue; + _nodeConnectionTracker = nodeConnectionTracker; } public async Task Read(CancellationToken cancellationToken = default) @@ -44,18 +44,25 @@ public async Task Read(CancellationToken cancellationToken = return QueueDashboardPage.Unavailable("Runtime statistics access was denied."); } + Task replicationConnectionsTask = null; try { using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); timeout.CancelAfter(ReadTimeout); - var queues = await ReadQueueStats(timeout.Token); - var tcp = await ReadTcpStatsSafe(timeout.Token, cancellationToken); - return QueueDashboardPage.Success(queues, tcp.Rows, tcp.Message); + var queuesTask = ReadQueueStats(timeout.Token); + replicationConnectionsTask = ReadReplicationStatsOrEmpty(timeout.Token, cancellationToken); + await Task.WhenAll(queuesTask, replicationConnectionsTask); + var replication = await replicationConnectionsTask; + return QueueDashboardPage.Success( + await queuesTask, + replication.Rows, + _nodeConnectionTracker.Snapshot(), + replication.Message); } catch (TimeoutException) { - return QueueDashboardPage.Unavailable("Timed out reading queue statistics."); + return QueueUnavailable("Timed out reading queue statistics."); } catch (OperationCanceledException) { @@ -64,11 +71,20 @@ public async Task Read(CancellationToken cancellationToken = throw; } - return QueueDashboardPage.Unavailable("Timed out reading queue statistics."); + return QueueUnavailable("Timed out reading queue statistics."); } catch (Exception ex) { - return QueueDashboardPage.Unavailable($"Unable to read queue statistics: {UiMessages.Friendly(ex)}"); + return QueueUnavailable($"Unable to read queue statistics: {UiMessages.Friendly(ex)}"); + } + + QueueDashboardPage QueueUnavailable(string message) + { + var replication = replicationConnectionsTask is { IsCompletedSuccessfully: true } + ? replicationConnectionsTask.Result + : new ReplicationStatsRead(Array.Empty(), ""); + return QueueDashboardPage.QueueUnavailable( + message, _nodeConnectionTracker.Snapshot(), replication.Rows, replication.Message); } } @@ -102,68 +118,55 @@ private async Task> ReadQueueStats(Cancellation return queues; } - private async Task ReadTcpStats(CancellationToken cancellationToken) + private async Task> ReadReplicationStats( + CancellationToken cancellationToken) { - if (!await HasAccess(TcpStatisticsOperation, cancellationToken)) - { - return new TcpConnectionResult(Array.Empty(), "TCP statistics access was denied."); - } - - var envelope = new TaskCompletionEnvelope(); - _monitoringQueue.Publish(new MonitoringMessage.GetFreshTcpConnectionStats(envelope)); + var envelope = new TaskCompletionEnvelope(); + _monitoringQueue.Publish(new ReplicationMessage.GetReplicationStats(envelope)); var completed = await envelope.Task.WaitAsync(ReadTimeout, cancellationToken); - return BuildTcpRows(completed.ConnectionStats ?? []); + + return completed.ReplicationStats + .Select(ReplicationConnectionRow.From) + .OrderBy(x => x.Endpoint, StringComparer.OrdinalIgnoreCase) + .ToArray(); } - private async Task ReadTcpStatsSafe( + private async Task ReadReplicationStatsOrEmpty( CancellationToken timeoutToken, CancellationToken cancellationToken) { try { - return await ReadTcpStats(timeoutToken); + if (!await HasAccess(ReplicationStatisticsOperation, timeoutToken)) + { + return new ReplicationStatsRead( + Array.Empty(), + "Replication statistics access was denied."); + } + + return new ReplicationStatsRead(await ReadReplicationStats(timeoutToken), ""); } - catch (TimeoutException) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - return new TcpConnectionResult(Array.Empty(), "Timed out reading TCP statistics."); + throw; } catch (OperationCanceledException) { - if (cancellationToken.IsCancellationRequested) - { - throw; - } - - return new TcpConnectionResult(Array.Empty(), "Timed out reading TCP statistics."); + return new ReplicationStatsRead( + Array.Empty(), + "Timed out reading replication statistics."); } catch (Exception ex) { - return new TcpConnectionResult( - Array.Empty(), - $"Unable to read TCP statistics: {UiMessages.Friendly(ex)}"); + return new ReplicationStatsRead( + Array.Empty(), + $"Unable to read replication statistics: {UiMessages.Friendly(ex)}"); } } - private TcpConnectionResult BuildTcpRows(IReadOnlyList connections) - { - lock (_tcpGate) - { - var now = DateTime.UtcNow; - var elapsedSeconds = _lastTcpRefresh.HasValue - ? Math.Max(1, (now - _lastTcpRefresh.Value).TotalSeconds) - : 1; - - var rows = connections - .Select(x => TcpConnectionRow.From(x, _previousTcpConnections.GetValueOrDefault(x.ConnectionId), elapsedSeconds)) - .OrderBy(x => x.ClientConnectionName, StringComparer.OrdinalIgnoreCase) - .ThenBy(x => x.ConnectionId) - .ToArray(); - - _previousTcpConnections = rows.ToDictionary(x => x.ConnectionId); - _lastTcpRefresh = now; - return new TcpConnectionResult(rows, ""); - } - } + private sealed record ReplicationStatsRead( + IReadOnlyList Rows, + string Message); } @@ -196,9 +199,11 @@ public static bool TryReadDictionary(object value, out IReadOnlyDictionary Blocks, IReadOnlyList Queues, - IReadOnlyList TcpConnections, - string TcpMessage, - string Message) + IReadOnlyList ReplicationConnections, + IReadOnlyList NodeConnections, + bool NetworkAvailable, + string Message, + string ReplicationMessage) { private static readonly JsonSerializerOptions PayloadJsonOptions = new(JsonSerializerDefaults.Web); @@ -214,24 +219,50 @@ public sealed record QueueDashboardPage( public string ClientPayloadJson => JsonSerializer.Serialize( new QueueDashboardPayload( Queues.Select(QueuePayload.From).ToArray(), - TcpConnections.Select(TcpConnectionPayload.From).ToArray(), + ReplicationConnections, + NodeConnections, + NetworkAvailable, Message, - TcpMessage), + ReplicationMessage), PayloadJsonOptions); public static QueueDashboardPage Success( IReadOnlyList queues, - IReadOnlyList tcpConnections, - string tcpMessage) => - new(BuildBlocks(queues), queues, tcpConnections, tcpMessage, ""); + IReadOnlyList replicationConnections = null, + IReadOnlyList nodeConnections = null, + string replicationMessage = "") => + new( + BuildBlocks(queues), + queues, + replicationConnections ?? Array.Empty(), + nodeConnections ?? Array.Empty(), + true, + "", + replicationMessage); + + public static QueueDashboardPage QueueUnavailable( + string message, + IReadOnlyList nodeConnections, + IReadOnlyList replicationConnections = null, + string replicationMessage = "") => + new( + Array.Empty(), + Array.Empty(), + replicationConnections ?? Array.Empty(), + nodeConnections, + true, + message, + replicationMessage); public static QueueDashboardPage Unavailable(string message) => new( Array.Empty(), Array.Empty(), - Array.Empty(), + Array.Empty(), + Array.Empty(), + false, message, - message); + ""); private static IReadOnlyList BuildBlocks(IReadOnlyList queues) { @@ -266,15 +297,35 @@ public sealed record QueueDashboardBlock( public bool HasChildren => Children.Count > 0; } -public sealed record TcpConnectionResult( - IReadOnlyList Rows, - string Message); - public sealed record QueueDashboardPayload( IReadOnlyList Queues, - IReadOnlyList TcpConnections, + IReadOnlyList ReplicationConnections, + IReadOnlyList NodeConnections, + bool NetworkAvailable, string Message, - string TcpMessage); + string ReplicationMessage); + +public sealed record ReplicationConnectionRow( + string SubscriptionId, + string ConnectionId, + string Endpoint, + long TotalBytesSent, + long TotalBytesReceived, + int PendingSendBytes, + int PendingReceivedBytes, + int SendQueueSize) +{ + public static ReplicationConnectionRow From(ReplicationMessage.ReplicationStats stats) => + new( + stats.SubscriptionId.ToString("D"), + stats.ConnectionId.ToString("D"), + stats.SubscriptionEndpoint ?? "", + stats.TotalBytesSent, + stats.TotalBytesReceived, + stats.PendingSendBytes, + stats.PendingReceivedBytes, + stats.SendQueueSize); +} public sealed record QueuePayload( string Kind, @@ -304,95 +355,6 @@ public static QueuePayload From(QueueDashboardRow row) => row.LastProcessedMessage); } -public sealed record TcpConnectionPayload( - Guid ConnectionId, - string ClientConnectionName, - string RemoteEndPoint, - string LocalEndPoint, - long TotalBytesSent, - long TotalBytesReceived, - int PendingSendBytes, - int PendingReceivedBytes, - double SentRate, - double ReceivedRate, - bool IsExternalConnection, - bool IsSslConnection) -{ - public static TcpConnectionPayload From(TcpConnectionRow row) => - new( - row.ConnectionId, - row.ClientConnectionName, - row.RemoteEndPoint, - row.LocalEndPoint, - row.TotalBytesSent, - row.TotalBytesReceived, - row.PendingSendBytes, - row.PendingReceivedBytes, - row.SentRate, - row.ReceivedRate, - row.IsExternalConnection, - row.IsSslConnection); -} - -public sealed record TcpConnectionRow( - Guid ConnectionId, - string ClientConnectionName, - string RemoteEndPoint, - string LocalEndPoint, - long TotalBytesSent, - long TotalBytesReceived, - int PendingSendBytes, - int PendingReceivedBytes, - double SentRate, - double ReceivedRate, - bool IsExternalConnection, - bool IsSslConnection) -{ - public string IdLabel => ConnectionId == Guid.Empty ? "" : ConnectionId.ToString("D"); - public string ClientLabel => DisplayMessage(ClientConnectionName); - public string TypeLabel => $"{(IsExternalConnection ? "External" : "Internal")} {(IsSslConnection ? "TLS" : "TCP")}"; - public string RemoteEndPointLabel => DisplayMessage(RemoteEndPoint); - public string SentRateLabel => FormatByteRate(SentRate); - public string ReceivedRateLabel => FormatByteRate(ReceivedRate); - public string TotalBytesSentLabel => TotalBytesSent.ToString("N0", CultureInfo.InvariantCulture); - public string TotalBytesReceivedLabel => TotalBytesReceived.ToString("N0", CultureInfo.InvariantCulture); - public string PendingSendBytesLabel => PendingSendBytes.ToString("N0", CultureInfo.InvariantCulture); - public string PendingReceivedBytesLabel => PendingReceivedBytes.ToString("N0", CultureInfo.InvariantCulture); - - public static TcpConnectionRow From( - MonitoringMessage.TcpConnectionStats stats, - TcpConnectionRow previous, - double elapsedSeconds) - { - var sentRate = previous is null - ? 0 - : Math.Max(0, (stats.TotalBytesSent - previous.TotalBytesSent) / elapsedSeconds); - var receivedRate = previous is null - ? 0 - : Math.Max(0, (stats.TotalBytesReceived - previous.TotalBytesReceived) / elapsedSeconds); - - return new TcpConnectionRow( - stats.ConnectionId, - stats.ClientConnectionName ?? "", - stats.RemoteEndPoint ?? "", - stats.LocalEndPoint ?? "", - stats.TotalBytesSent, - stats.TotalBytesReceived, - stats.PendingSendBytes, - stats.PendingReceivedBytes, - sentRate, - receivedRate, - stats.IsExternalConnection, - stats.IsSslConnection); - } - - private static string DisplayMessage(string value) => - string.IsNullOrWhiteSpace(value) ? "" : value; - - private static string FormatByteRate(double value) => - $"{Math.Round(value).ToString("N0", CultureInfo.InvariantCulture)} B/s"; -} - public enum QueueDashboardRowKind { Queue, diff --git a/src/EventStore.ClusterNode/Program.cs b/src/EventStore.ClusterNode/Program.cs index e1127f64e2..d6139e1ae0 100644 --- a/src/EventStore.ClusterNode/Program.cs +++ b/src/EventStore.ClusterNode/Program.cs @@ -20,6 +20,7 @@ using EventStore.Core.Authentication.OAuth; using EventStore.Core.Certificates; using EventStore.Core.Configuration; +using EventStore.Core.Services.Transport.Grpc; using EventStore.Core.Services.Transport.Http; using EventStore.Plugins.Authentication; using Microsoft.AspNetCore.Builder; @@ -273,6 +274,9 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig { x.SuppressStatusMessages = true; }); + var nodeConnectionTracker = new NodeConnectionTracker(); + builder.Services.AddSingleton(nodeConnectionTracker); + builder.Services.AddSingleton(nodeConnectionTracker); EndpointBinding[] endpointBindings = [ new(EndpointRole.Client, @@ -292,7 +296,6 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig ], defaultRouteRole: EndpointRole.Client, nonIpEndpointRole: EndpointRole.Client); - builder.WebHost.ConfigureKestrel(server => { server.Limits.Http2.KeepAlivePingDelay = @@ -303,14 +306,14 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig foreach (var binding in endpointBindings) { server.Listen(binding.ListenEndPoint, listenOptions => - ConfigureHttpOptions(listenOptions, hostedService, + ConfigureHttpOptions(listenOptions, hostedService, nodeConnectionTracker, useHttps: !hostedService.Node.DisableHttps, protocols: binding.Protocols)); } if (hostedService.Node.EnableUnixSocket) { - TryListenOnUnixSocket(hostedService, server); + TryListenOnUnixSocket(hostedService, server, nodeConnectionTracker); } }); @@ -349,6 +352,19 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig builder.Services.AddSingleton(hostedService); var app = builder.Build(); + app.Use((context, next) => + { + var isGrpc = context.Request.ContentType?.StartsWith( + "application/grpc", + StringComparison.OrdinalIgnoreCase) == true; + nodeConnectionTracker.ObserveRequest( + context.Connection.Id, + context.Request.Protocol, + isGrpc, + context.Request.Headers["connection-name"].FirstOrDefault(), + context.Request.Headers.UserAgent.ToString()); + return next(context); + }); app.Use(async (context, next) => { if (!endpointPolicy.Allows(context)) @@ -413,9 +429,11 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig private static void ConfigureHttpOptions( ListenOptions listenOptions, ClusterVNodeHostedService hostedService, + NodeConnectionTracker connectionTracker, bool useHttps, HttpProtocols protocols = HttpProtocols.Http1AndHttp2) { + listenOptions.Use(next => context => connectionTracker.Track(context, next, useHttps)); listenOptions.Protocols = protocols; if (useHttps) @@ -429,7 +447,10 @@ private static void ConfigureHttpOptions( } } - private static void TryListenOnUnixSocket(ClusterVNodeHostedService hostedService, KestrelServerOptions server) + private static void TryListenOnUnixSocket( + ClusterVNodeHostedService hostedService, + KestrelServerOptions server, + NodeConnectionTracker connectionTracker) { if (!RuntimeInformation.IsLinux && !OperatingSystem.IsWindowsVersionAtLeast(10, 0, 17063)) { @@ -460,7 +481,7 @@ private static void TryListenOnUnixSocket(ClusterVNodeHostedService hostedServic server.ListenUnixSocket(unixSocket, listenOptions => { listenOptions.Use(next => new UnixSocketConnectionMiddleware(next).OnConnectAsync); - ConfigureHttpOptions(listenOptions, hostedService, useHttps: false); + ConfigureHttpOptions(listenOptions, hostedService, connectionTracker, useHttps: false); }); Log.Information("Listening on UNIX domain socket: {unixSocket}", unixSocket); } diff --git a/src/EventStore.ClusterNode/metricsconfig.json b/src/EventStore.ClusterNode/metricsconfig.json index cb1edb93ff..f249180912 100644 --- a/src/EventStore.ClusterNode/metricsconfig.json +++ b/src/EventStore.ClusterNode/metricsconfig.json @@ -236,10 +236,6 @@ "Regex": "CoreMessage-System-.*", "Label": "System" }, - { - "Regex": "CoreMessage-Tcp-.*", - "Label": "Tcp" - }, { "Regex": "ProjectionMessage-.*", "Label": "Projections" diff --git a/src/EventStore.ClusterNode/ui-assets/js/queue-dashboard.js b/src/EventStore.ClusterNode/ui-assets/js/queue-dashboard.js index d4602e7bec..108f020cc8 100644 --- a/src/EventStore.ClusterNode/ui-assets/js/queue-dashboard.js +++ b/src/EventStore.ClusterNode/ui-assets/js/queue-dashboard.js @@ -2,7 +2,7 @@ "use strict"; var pollIntervalMs = 1000; - var tcpPageSize = 5; + var networkPageSize = 5; var dashboards = new WeakMap(); function start() { @@ -20,8 +20,11 @@ expanded: parseExpanded(root), blocks: [], queues: [], - tcpConnections: [], - tcpPage: 0, + networkConnections: [], + networkAvailable: false, + networkSamples: new Map(), + networkPage: 0, + replicationConnections: [], timer: null, inFlight: false }; @@ -33,10 +36,10 @@ applyPayload(state, initialPayload); root.addEventListener("click", function (event) { - var tcpPager = event.target.closest("[data-tcp-page]"); - if (tcpPager && root.contains(tcpPager)) { + var networkPager = event.target.closest("[data-network-page]"); + if (networkPager && root.contains(networkPager)) { event.preventDefault(); - changeTcpPage(state, tcpPager.getAttribute("data-tcp-page")); + changeNetworkPage(state, networkPager.getAttribute("data-network-page")); return; } @@ -124,21 +127,18 @@ var parsed = parseQueues(payload); state.queues = parsed.queues; state.blocks = parsed.blocks; - state.tcpConnections = parseTcpConnections(payload); + state.networkConnections = parseNetworkConnections(payload, state); + state.networkAvailable = payload.networkAvailable === true; + state.replicationConnections = parseReplicationConnections(payload); setStatus( state.root, payload.message ? "Live stats unavailable" : "Live stats", payload.message || "Updated " + formatTime(new Date())); + setNetworkStatus( + state.root, + state.networkAvailable ? "" : (payload.message || "Network statistics are unavailable."), + state.networkConnections.length); render(state); - - if (state.root.querySelector("[data-tcp-table-body]")) { - var tcpMessage = payload.tcpMessage || payload.message || ""; - setTcpStatus( - state.root, - tcpMessage ? "TCP unavailable" : "TCP live", - tcpMessage || state.tcpConnections.length + " connection" + (state.tcpConnections.length === 1 ? "" : "s")); - renderTcpTable(state.root, state.tcpConnections, state); - } } async function refresh(state) { @@ -159,11 +159,12 @@ } catch (error) { state.queues = []; state.blocks = []; - state.tcpConnections = []; + state.networkConnections = []; + state.networkAvailable = false; + state.replicationConnections = []; setStatus(state.root, "Live stats unavailable", friendlyMessage(error)); - setTcpStatus(state.root, "TCP unavailable", friendlyMessage(error)); + setNetworkStatus(state.root, friendlyMessage(error), 0); render(state); - renderTcpTable(state.root, state.tcpConnections, state); } finally { state.inFlight = false; } @@ -208,32 +209,72 @@ }; } - function parseTcpConnections(payload) { - var rows = payload && Array.isArray(payload.tcpConnections) ? payload.tcpConnections : []; - - return rows.map(function (row) { - var id = readFieldString(row, ["connectionId", "ConnectionId"], ""); - var totalBytesSent = readFieldNumber(row, ["totalBytesSent", "TotalBytesSent"]); - var totalBytesReceived = readFieldNumber(row, ["totalBytesReceived", "TotalBytesReceived"]); + function parseNetworkConnections(payload, state) { + var rows = payload && Array.isArray(payload.nodeConnections) ? payload.nodeConnections : []; + var now = Date.now(); + var nextSamples = new Map(); + var connections = rows.map(function (row) { + var id = readString(row.connectionId, ""); + var totalBytesSent = readNumber(row.totalBytesSent); + var totalBytesReceived = readNumber(row.totalBytesReceived); + var previous = state.networkSamples.get(id); + var elapsedSeconds = previous ? Math.max(0.001, (now - previous.observedAt) / 1000) : 0; + var sentRate = previous ? Math.max(0, totalBytesSent - previous.totalBytesSent) / elapsedSeconds : 0; + var receivedRate = previous + ? Math.max(0, totalBytesReceived - previous.totalBytesReceived) / elapsedSeconds + : 0; + + nextSamples.set(id, { + observedAt: now, + totalBytesSent: totalBytesSent, + totalBytesReceived: totalBytesReceived + }); return { id: id, - clientConnectionName: readFieldString(row, ["clientConnectionName", "ClientConnectionName"], ""), - remoteEndPoint: readFieldString(row, ["remoteEndPoint", "RemoteEndPoint"], ""), - localEndPoint: readFieldString(row, ["localEndPoint", "LocalEndPoint"], ""), + clientName: readString(row.clientName, ""), + application: readString(row.application, "Awaiting request"), + protocol: readString(row.protocol, ""), + remoteEndPoint: readString(row.remoteEndPoint, ""), + localEndPoint: readString(row.localEndPoint, ""), + connectedAt: readString(row.connectedAt, ""), + isTls: Boolean(row.isTls), totalBytesSent: totalBytesSent, totalBytesReceived: totalBytesReceived, - pendingSendBytes: readFieldNumber(row, ["pendingSendBytes", "PendingSendBytes"]), - pendingReceivedBytes: readFieldNumber(row, ["pendingReceivedBytes", "PendingReceivedBytes"]), - sentRate: readFieldNumber(row, ["sentRate", "SentRate"]), - receivedRate: readFieldNumber(row, ["receivedRate", "ReceivedRate"]), - isExternalConnection: readFieldBoolean(row, ["isExternalConnection", "IsExternalConnection"]), - isSslConnection: readFieldBoolean(row, ["isSslConnection", "IsSslConnection"]) + pendingSendBytes: readNumber(row.pendingSendBytes), + pendingReceivedBytes: readNumber(row.pendingReceivedBytes), + sentRate: sentRate, + receivedRate: receivedRate }; }).sort(function (left, right) { - return left.clientConnectionName.localeCompare(right.clientConnectionName, undefined, { sensitivity: "base" }) || + return left.clientName.localeCompare(right.clientName, undefined, { sensitivity: "base" }) || left.id.localeCompare(right.id, undefined, { sensitivity: "base" }); }); + + state.networkSamples = nextSamples; + return connections; + } + + function parseReplicationConnections(payload) { + var rows = payload && Array.isArray(payload.replicationConnections) + ? payload.replicationConnections + : []; + + return rows.map(function (row) { + return { + subscriptionId: readString(row.subscriptionId, ""), + connectionId: readString(row.connectionId, ""), + endpoint: readString(row.endpoint, ""), + totalBytesSent: readNumber(row.totalBytesSent), + totalBytesReceived: readNumber(row.totalBytesReceived), + pendingSendBytes: readNumber(row.pendingSendBytes), + pendingReceivedBytes: readNumber(row.pendingReceivedBytes), + sendQueueSize: readNumber(row.sendQueueSize) + }; + }).sort(function (left, right) { + return left.endpoint.localeCompare(right.endpoint, undefined, { sensitivity: "base" }) || + left.connectionId.localeCompare(right.connectionId, undefined, { sensitivity: "base" }); + }); } function buildBlocks(queues) { @@ -311,6 +352,8 @@ renderSpotlightTable(state.root, state.queues); renderQueueTable(state); renderDashboardSnapshot(state.root, state.blocks); + renderNetworkTable(state); + renderReplicationTable(state.root, state.replicationConnections); } function updateMetrics(root, queues) { @@ -451,32 +494,35 @@ node.textContent = lines.join("\n"); } - function renderTcpTable(root, connections, state) { - var tbody = root.querySelector("[data-tcp-table-body]"); + function renderNetworkTable(state) { + var tbody = state.root.querySelector("[data-network-table-body]"); if (!tbody) return; replaceChildren(tbody); - if (connections.length === 0) { + if (state.networkConnections.length === 0) { var empty = element("tr"); - var cell = element("td", "px-5 py-4 text-es-muted"); - cell.colSpan = 10; - cell.textContent = "No TCP connections are currently reported."; - empty.appendChild(cell); + var emptyCell = element("td", "px-5 py-4 text-es-muted"); + emptyCell.colSpan = 10; + emptyCell.textContent = state.networkAvailable + ? "No active shared-endpoint connections." + : "Network statistics are unavailable."; + empty.appendChild(emptyCell); tbody.appendChild(empty); - updateTcpPagination(root, state, 0); + updateNetworkPagination(state, 0); return; } - var pageCount = Math.ceil(connections.length / tcpPageSize); - state.tcpPage = Math.min(state.tcpPage, Math.max(0, pageCount - 1)); - var pageStart = state.tcpPage * tcpPageSize; - connections.slice(pageStart, pageStart + tcpPageSize).forEach(function (connection) { + var pageCount = Math.ceil(state.networkConnections.length / networkPageSize); + state.networkPage = Math.min(state.networkPage, pageCount - 1); + var offset = state.networkPage * networkPageSize; + state.networkConnections.slice(offset, offset + networkPageSize).forEach(function (connection) { var row = element("tr", "bg-white/70 text-es-ink"); - appendText(row, "td", connection.id || "", "max-w-[14rem] truncate px-5 py-4 font-mono text-xs text-es-muted"); - appendText(row, "td", displayMessage(connection.clientConnectionName), "px-5 py-4 font-bold text-es-ink"); - appendText(row, "td", tcpTypeLabel(connection), "px-5 py-4 text-es-muted"); - appendText(row, "td", displayMessage(connection.remoteEndPoint), "px-5 py-4 font-mono text-xs text-es-muted"); + appendText(row, "td", displayMessage(connection.id), "max-w-[14rem] truncate px-5 py-4 font-mono text-xs text-es-muted"); + appendText(row, "td", displayMessage(connection.clientName), "max-w-[16rem] truncate px-5 py-4 font-bold text-es-ink"); + appendText(row, "td", networkTypeLabel(connection), "px-5 py-4 text-es-muted"); + var endpointCell = appendText(row, "td", displayMessage(connection.remoteEndPoint), "px-5 py-4 font-mono text-xs text-es-muted"); + endpointCell.title = networkConnectionDetails(connection); appendText(row, "td", formatByteRate(connection.sentRate), "px-5 py-4 text-right font-mono text-es-ink"); appendText(row, "td", formatInteger(connection.totalBytesSent), "px-5 py-4 text-right font-mono text-es-ink"); appendText(row, "td", formatInteger(connection.pendingSendBytes), "px-5 py-4 text-right font-mono text-es-ink"); @@ -485,17 +531,80 @@ appendText(row, "td", formatInteger(connection.pendingReceivedBytes), "px-5 py-4 text-right font-mono text-es-ink"); tbody.appendChild(row); }); - updateTcpPagination(root, state, pageCount); + + updateNetworkPagination(state, pageCount); } - function changeTcpPage(state, direction) { - var pageCount = Math.ceil(state.tcpConnections.length / tcpPageSize); + function renderReplicationTable(root, connections) { + var tbody = root.querySelector("[data-replication-table-body]"); + if (!tbody) + return; + + replaceChildren(tbody); + if (connections.length === 0) { + var empty = element("tr"); + var emptyCell = element("td", "px-5 py-4 text-es-muted"); + emptyCell.colSpan = 6; + emptyCell.textContent = "No active gRPC replication connections."; + empty.appendChild(emptyCell); + tbody.appendChild(empty); + return; + } + + connections.forEach(function (connection) { + var row = element("tr", "bg-white/70 text-es-ink"); + appendText(row, "td", connection.endpoint, "px-5 py-4 font-bold text-es-ink"); + var connectionCell = appendText(row, "td", displayMessage(connection.connectionId), "px-5 py-4 font-mono text-xs text-es-muted"); + connectionCell.title = "Subscription " + displayMessage(connection.subscriptionId); + appendText(row, "td", formatInteger(connection.totalBytesSent), "px-5 py-4 text-right font-mono text-es-ink"); + appendText(row, "td", formatInteger(connection.totalBytesReceived), "px-5 py-4 text-right font-mono text-es-ink"); + appendText(row, "td", formatInteger(connection.pendingSendBytes + connection.pendingReceivedBytes), "px-5 py-4 text-right font-mono text-es-ink"); + appendText(row, "td", formatInteger(connection.sendQueueSize), "px-5 py-4 text-right font-mono text-es-ink"); + tbody.appendChild(row); + }); + } + + function changeNetworkPage(state, direction) { + var pageCount = Math.ceil(state.networkConnections.length / networkPageSize); if (direction === "previous") - state.tcpPage = Math.max(0, state.tcpPage - 1); + state.networkPage = Math.max(0, state.networkPage - 1); else if (direction === "next") - state.tcpPage = Math.min(Math.max(0, pageCount - 1), state.tcpPage + 1); + state.networkPage = Math.min(Math.max(0, pageCount - 1), state.networkPage + 1); - renderTcpTable(state.root, state.tcpConnections, state); + renderNetworkTable(state); + } + + function updateNetworkPagination(state, pageCount) { + var status = state.root.querySelector("[data-network-page-status]"); + var previous = state.root.querySelector('[data-network-page="previous"]'); + var next = state.root.querySelector('[data-network-page="next"]'); + var hasRows = state.networkConnections.length > 0; + + if (status) + status.textContent = hasRows ? "Page " + (state.networkPage + 1) + " of " + pageCount : "No pages"; + if (previous) + previous.disabled = !hasRows || state.networkPage === 0; + if (next) + next.disabled = !hasRows || state.networkPage >= pageCount - 1; + } + + function setNetworkStatus(root, message, connectionCount) { + var status = root.querySelector("[data-network-status]"); + if (!status) + return; + + status.textContent = message + ? "Network unavailable · " + message + : "Network live · " + connectionCount + " connection" + (connectionCount === 1 ? "" : "s"); + } + + function networkTypeLabel(connection) { + return connection.application + " · " + connection.protocol + " · " + (connection.isTls ? "TLS" : "Cleartext"); + } + + function networkConnectionDetails(connection) { + var connected = connection.connectedAt ? new Date(connection.connectedAt).toLocaleString() : "unknown"; + return "Local " + connection.localEndPoint + ", connected " + connected; } function queueTableRow(queue, block, state) { @@ -554,37 +663,6 @@ updatedNode.textContent = updated; } - function setTcpStatus(root, status, detail) { - var statusNode = root.querySelector("[data-tcp-status]"); - if (!statusNode) - return; - - statusNode.textContent = detail ? status + " · " + detail : status; - } - - function updateTcpPagination(root, state, pageCount) { - var status = root.querySelector("[data-tcp-page-status]"); - var previous = root.querySelector('[data-tcp-page="previous"]'); - var next = root.querySelector('[data-tcp-page="next"]'); - var hasRows = state.tcpConnections.length > 0; - - if (status) - status.textContent = hasRows - ? "Page " + (state.tcpPage + 1) + " of " + pageCount - : "No pages"; - - if (previous) - previous.disabled = !hasRows || state.tcpPage === 0; - - if (next) - next.disabled = !hasRows || state.tcpPage >= pageCount - 1; - } - - function tcpTypeLabel(connection) { - return (connection.isExternalConnection ? "External" : "Internal") + " " + - (connection.isSslConnection ? "TLS" : "TCP"); - } - function currentLastMessage(queue) { return queue.kind === "group" ? "n/a" @@ -662,36 +740,6 @@ return Number.isFinite(number) ? number : 0; } - function readFieldString(source, keys, fallback) { - var value = readField(source, keys); - if (value === null || value === undefined) - return fallback; - - var text = String(value); - return text.trim() ? text : fallback; - } - - function readFieldNumber(source, keys) { - return readNumber(readField(source, keys)); - } - - function readFieldBoolean(source, keys) { - var value = readField(source, keys); - return value === true || String(value).toLowerCase() === "true"; - } - - function readField(source, keys) { - if (!source || typeof source !== "object") - return undefined; - - for (var i = 0; i < keys.length; i++) { - if (Object.prototype.hasOwnProperty.call(source, keys[i])) - return source[keys[i]]; - } - - return undefined; - } - function sum(rows, key) { return rows.reduce(function (total, row) { return total + row[key]; diff --git a/src/EventStore.Core.Tests/Regression/GrpcOnlySurfaceParityTests.cs b/src/EventStore.Core.Tests/Regression/GrpcOnlySurfaceParityTests.cs new file mode 100644 index 0000000000..40bbe69f75 --- /dev/null +++ b/src/EventStore.Core.Tests/Regression/GrpcOnlySurfaceParityTests.cs @@ -0,0 +1,330 @@ +using System; +using System.Collections.Generic; +using System.IO.Pipelines; +using System.Net; +using System.Security.Claims; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using EventStore.ClusterNode.Components.Services; +using EventStore.Core; +using EventStore.Core.Authorization; +using EventStore.Core.Bus; +using EventStore.Core.Messages; +using EventStore.Core.Messaging; +using EventStore.Plugins.Authorization; +using Microsoft.AspNetCore.Connections; +using Microsoft.AspNetCore.Http; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Regression; + +[TestFixture] +public class GrpcOnlySurfaceParityTests +{ + [Test] + public void observability_payload_preserves_replication_visibility() + { + var page = QueueDashboardPage.Success(Array.Empty()); + using var payload = JsonDocument.Parse(page.ClientPayloadJson); + + Assert.That(payload.RootElement.TryGetProperty("replicationConnections", out _), Is.True); + Assert.That(payload.RootElement.TryGetProperty("nodeConnections", out _), Is.True); + } + + [Test] + public async Task replication_stats_failure_does_not_hide_queue_statistics() + { + var publisher = new QueueStatsPublisher(); + var components = new StandardComponents( + null, null, null, null, null, null, null, publisher, null, null, false); + var service = new QueueDashboardService( + new PassthroughAuthorizationProvider(), + new HttpContextAccessor { HttpContext = new DefaultHttpContext() }, + components, + new NodeConnectionTracker()); + + var page = await service.Read(); + + Assert.Multiple(() => + { + Assert.That(page.IsAvailable, Is.True); + Assert.That(page.Queues, Has.One.Matches(x => x.Name == "mainQueue")); + Assert.That(page.ReplicationMessage, Does.StartWith("Unable to read replication statistics:")); + }); + } + + [Test] + public async Task queue_stats_failure_does_not_hide_active_connections() + { + var (tracker, release, tracking) = TrackActiveConnection(); + + try + { + var components = new StandardComponents( + null, null, null, null, null, null, null, new QueueStatsPublisher(failQueueStats: true), + null, null, false); + var service = new QueueDashboardService( + new PassthroughAuthorizationProvider(), + new HttpContextAccessor { HttpContext = new DefaultHttpContext() }, + components, + tracker); + + var page = await service.Read(); + using var payload = JsonDocument.Parse(page.ClientPayloadJson); + + Assert.Multiple(() => + { + Assert.That(page.IsAvailable, Is.False); + Assert.That(page.NodeConnections, Has.One.Matches(x => + x.ConnectionId == "active-connection")); + Assert.That(payload.RootElement.GetProperty("nodeConnections").GetArrayLength(), Is.EqualTo(1)); + Assert.That(payload.RootElement.TryGetProperty("networkAvailable", out var networkAvailable), Is.True); + Assert.That(networkAvailable.GetBoolean(), Is.True); + }); + } + finally + { + release.SetResult(); + await tracking; + } + } + + [Test] + public async Task queue_stats_failure_does_not_hide_replication_connections() + { + var components = new StandardComponents( + null, null, null, null, null, null, null, + new QueueStatsPublisher(failQueueStats: true, provideReplicationStats: true), + null, null, false); + var service = new QueueDashboardService( + new PassthroughAuthorizationProvider(), + new HttpContextAccessor { HttpContext = new DefaultHttpContext() }, + components, + new NodeConnectionTracker()); + + var page = await service.Read(); + using var payload = JsonDocument.Parse(page.ClientPayloadJson); + + Assert.Multiple(() => + { + Assert.That(page.IsAvailable, Is.False); + Assert.That(page.ReplicationConnections, + Has.One.Matches(x => x.Endpoint == "replica:1112")); + Assert.That(payload.RootElement.GetProperty("replicationConnections").GetArrayLength(), Is.EqualTo(1)); + }); + } + + [Test] + public async Task replication_statistics_require_replication_access_without_hiding_other_diagnostics() + { + var (tracker, release, tracking) = TrackActiveConnection(); + try + { + var publisher = new QueueStatsPublisher(provideReplicationStats: true); + var authorization = new ReadOnlyStatisticsAuthorizationProvider(); + var components = new StandardComponents( + null, null, null, null, null, null, null, publisher, null, null, false); + var service = new QueueDashboardService( + authorization, + new HttpContextAccessor { HttpContext = new DefaultHttpContext() }, + components, + tracker); + + var page = await service.Read(); + using var payload = JsonDocument.Parse(page.ClientPayloadJson); + + Assert.Multiple(() => + { + Assert.That(page.IsAvailable, Is.True); + Assert.That(page.Queues, Has.One.Matches(x => x.Name == "mainQueue")); + Assert.That(page.NodeConnections, Has.One.Matches(x => + x.ConnectionId == "active-connection")); + Assert.That(page.ReplicationConnections, Is.Empty); + Assert.That(page.ReplicationMessage, Is.EqualTo("Replication statistics access was denied.")); + Assert.That(payload.RootElement.GetProperty("replicationConnections").GetArrayLength(), Is.Zero); + Assert.That(publisher.ReplicationRequests, Is.Zero); + Assert.That(authorization.RequestedOperations, Is.EquivalentTo(new[] + { + new Operation(Operations.Node.Statistics.Read), + new Operation(Operations.Node.Statistics.Replication) + })); + }); + } + finally + { + release.SetResult(); + await tracking; + } + } + + [Test] + public async Task denied_statistics_access_does_not_expose_or_mark_connections_available() + { + var (tracker, release, tracking) = TrackActiveConnection(); + try + { + var components = new StandardComponents( + null, null, null, null, null, null, null, new QueueStatsPublisher(), + null, null, false); + var service = new QueueDashboardService( + new DenyingAuthorizationProvider(), + new HttpContextAccessor { HttpContext = new DefaultHttpContext() }, + components, + tracker); + + var page = await service.Read(); + using var payload = JsonDocument.Parse(page.ClientPayloadJson); + + Assert.Multiple(() => + { + Assert.That(page.IsAvailable, Is.False); + Assert.That(page.NodeConnections, Is.Empty); + Assert.That(payload.RootElement.GetProperty("nodeConnections").GetArrayLength(), Is.Zero); + Assert.That(payload.RootElement.TryGetProperty("networkAvailable", out var networkAvailable), Is.True); + Assert.That(networkAvailable.GetBoolean(), Is.False); + }); + } + finally + { + release.SetResult(); + await tracking; + } + } + + [Test] + public async Task http_connections_are_visible_only_while_active() + { + var tracker = new NodeConnectionTracker(); + var connection = new DefaultConnectionContext("grpc-connection") + { + LocalEndPoint = new IPEndPoint(IPAddress.Loopback, 2113), + RemoteEndPoint = new IPEndPoint(IPAddress.Loopback, 50123) + }; + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var trafficObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var incoming = new Pipe(); + var outgoing = new Pipe(); + connection.Transport = new TestDuplexPipe(incoming.Reader, outgoing.Writer); + var tracking = tracker.Track(connection, async trackedConnection => + { + tracker.ObserveRequest( + trackedConnection.ConnectionId, + "HTTP/2", + isGrpc: true, + connectionName: "projection-catchup", + userAgent: "grpc-dotnet"); + await trackedConnection.Transport.Output.WriteAsync(new byte[3]); + await incoming.Writer.WriteAsync(new byte[5]); + var read = await trackedConnection.Transport.Input.ReadAsync(); + trackedConnection.Transport.Input.AdvanceTo(read.Buffer.GetPosition(2), read.Buffer.End); + trafficObserved.SetResult(); + await release.Task; + }, isTls: true); + await trafficObserved.Task; + + Assert.That(tracker.Snapshot(), Has.One.Matches(x => + x.ConnectionId == "grpc-connection" && x.IsTls && + x.ClientName == "projection-catchup" && x.Application == "gRPC" && x.Protocol == "HTTP/2" && + x.TotalBytesSent == 3 && x.TotalBytesReceived == 2 && x.PendingReceivedBytes == 3)); + + tracker.ObserveRequest( + connection.ConnectionId, + "HTTP/2", + isGrpc: false, + connectionName: "", + userAgent: "browser"); + Assert.That(tracker.Snapshot(), Has.One.Matches(x => + x.ClientName == "projection-catchup" && x.Application == "HTTP and gRPC")); + + release.SetResult(); + await tracking; + + Assert.That(tracker.Snapshot(), Is.Empty); + } + + private sealed class TestDuplexPipe(PipeReader input, PipeWriter output) : IDuplexPipe + { + public PipeReader Input { get; } = input; + public PipeWriter Output { get; } = output; + } + + private static (NodeConnectionTracker Tracker, TaskCompletionSource Release, Task Tracking) + TrackActiveConnection() + { + var tracker = new NodeConnectionTracker(); + var connection = new DefaultConnectionContext("active-connection") + { + LocalEndPoint = new IPEndPoint(IPAddress.Loopback, 2113), + RemoteEndPoint = new IPEndPoint(IPAddress.Loopback, 50123), + Transport = new TestDuplexPipe(new Pipe().Reader, new Pipe().Writer) + }; + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var tracking = tracker.Track(connection, _ => release.Task, isTls: false); + return (tracker, release, tracking); + } + + private sealed class DenyingAuthorizationProvider : PassthroughAuthorizationProvider + { + public override ValueTask CheckAccessAsync( + ClaimsPrincipal principal, Operation operation, CancellationToken cancellationToken) => + ValueTask.FromResult(false); + } + + private sealed class ReadOnlyStatisticsAuthorizationProvider : PassthroughAuthorizationProvider + { + public List RequestedOperations { get; } = new(); + + public override ValueTask CheckAccessAsync( + ClaimsPrincipal principal, Operation operation, CancellationToken cancellationToken) + { + RequestedOperations.Add(operation); + return ValueTask.FromResult(!operation.Equals(new Operation(Operations.Node.Statistics.Replication))); + } + } + + private sealed class QueueStatsPublisher(bool failQueueStats = false, bool provideReplicationStats = false) : IPublisher + { + public int ReplicationRequests { get; private set; } + + public void Publish(Message message) + { + switch (message) + { + case MonitoringMessage.GetFreshStats request: + if (failQueueStats) + { + throw new InvalidOperationException("Queue statistics are unavailable."); + } + + request.Envelope.ReplyWith(new MonitoringMessage.GetFreshStatsCompleted( + success: true, + stats: new Dictionary + { + ["es"] = new Dictionary + { + ["queue"] = new Dictionary + { + ["mainQueue"] = new Dictionary + { + ["queueName"] = "mainQueue" + } + } + } + })); + break; + case ReplicationMessage.GetReplicationStats request: + ReplicationRequests++; + if (!provideReplicationStats) + { + throw new InvalidOperationException("Replication statistics are unavailable."); + } + + request.Envelope.ReplyWith(new ReplicationMessage.GetReplicationStatsCompleted( + [new ReplicationMessage.ReplicationStats( + Guid.NewGuid(), Guid.NewGuid(), "replica:1112", 0, 0, 0, 0, 0)])); + break; + } + } + } +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/AllowMonitoringAuthorizationProvider.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/AllowMonitoringAuthorizationProvider.cs new file mode 100644 index 0000000000..d69ca7838a --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/AllowMonitoringAuthorizationProvider.cs @@ -0,0 +1,12 @@ +using System.Security.Claims; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Plugins.Authorization; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; + +internal sealed class AllowMonitoringAuthorizationProvider : AuthorizationProviderBase +{ + public override ValueTask CheckAccessAsync(ClaimsPrincipal principal, Operation operation, + CancellationToken cancellationToken) => ValueTask.FromResult(true); +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsKestrelTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsKestrelTests.cs new file mode 100644 index 0000000000..f4a6620e0e --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsKestrelTests.cs @@ -0,0 +1,124 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Client.Monitoring; +using EventStore.ClusterNode.Components.Services; +using EventStore.Core.Bus; +using EventStore.Core.Messaging; +using EventStore.Core.Services.Transport.Grpc; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; + +[TestFixture] +public class ConnectionStatsKestrelTests +{ + [Test] + public async Task grpc_connection_is_visible_while_active_and_removed_after_disconnect() + { + var tracker = new NodeConnectionTracker(); + using var host = new HostBuilder() + .ConfigureWebHost(webHost => webHost + .UseKestrel(server => server.Listen(IPAddress.Loopback, 0, listenOptions => + { + listenOptions.Protocols = HttpProtocols.Http2; + listenOptions.Use(next => context => tracker.Track(context, next, isTls: false)); + })) + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(tracker); + }) + .Configure(app => + { + app.UseRouting(); + app.Use((context, next) => + { + tracker.ObserveRequest( + context.Connection.Id, + context.Request.Protocol, + context.Request.ContentType?.StartsWith("application/grpc", StringComparison.OrdinalIgnoreCase) == true, + context.Request.Headers["connection-name"].FirstOrDefault(), + context.Request.Headers.UserAgent.ToString()); + return next(); + }); + app.UseEndpoints(endpoints => endpoints.MapGrpcService()); + })) + .Build(); + + await host.StartAsync(); + var address = host.Services.GetRequiredService() + .Features.Get()!.Addresses.Single(); + string connectionId; + + using (var handler = new SocketsHttpHandler()) + using (var httpClient = new HttpClient(handler)) + using (var channel = GrpcChannel.ForAddress(address, new GrpcChannelOptions { HttpClient = httpClient })) + { + var client = new EventStore.Client.Monitoring.Monitoring.MonitoringClient(channel); + var response = await client.ConnectionStatsAsync( + new ConnectionStatsReq(), deadline: DateTime.UtcNow.AddSeconds(10)); + var connection = response.Connections.Single(x => x.Application == "gRPC"); + connectionId = connection.ConnectionId; + + Assert.Multiple(() => + { + Assert.That(connection.Protocol, Is.EqualTo("HTTP/2")); + Assert.That(connection.IsTls, Is.False); + Assert.That(connection.TotalBytesReceived, Is.GreaterThan(0)); + Assert.That(connection.RemoteEndpoint, Is.Not.Empty); + Assert.That(connection.LocalEndpoint, Is.Not.Empty); + Assert.That(tracker.Snapshot().Any(x => x.ConnectionId == connectionId), Is.True); + }); + } + + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + while (tracker.Snapshot().Any(x => x.ConnectionId == connectionId)) + { + await Task.Delay(25, timeout.Token); + } + } + + public sealed class ProductionMonitoringAdapter : EventStore.Client.Monitoring.Monitoring.MonitoringBase + { + private readonly object _service; + private readonly MethodInfo _method; + + public ProductionMonitoringAdapter(NodeConnectionTracker tracker) + { + var serviceType = typeof(IConnectionStatsProvider).Assembly.GetType( + "EventStore.Core.Services.Transport.Grpc.Monitoring", throwOnError: true)!; + _service = Activator.CreateInstance( + serviceType, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + binder: null, + args: [new RejectingPublisher(), tracker, new AllowMonitoringAuthorizationProvider()], + culture: null)!; + _method = serviceType.GetMethod(nameof(ConnectionStats))!; + } + + public override Task ConnectionStats( + ConnectionStatsReq request, ServerCallContext context) => + (Task)_method.Invoke(_service, [request, context])!; + } + + private sealed class RejectingPublisher : IPublisher + { + public void Publish(Message message) => + throw new InvalidOperationException($"Unexpected message {message.GetType().Name}"); + } +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsTests.cs new file mode 100644 index 0000000000..73995ead7c --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ConnectionStatsTests.cs @@ -0,0 +1,116 @@ +using System; +using System.Collections.Generic; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Client.Monitoring; +using EventStore.Core.Bus; +using EventStore.Core.Messaging; +using EventStore.Core.Services.Transport.Grpc; +using Grpc.Core; +using Microsoft.AspNetCore.Http; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; + +[TestFixture] +public class ConnectionStatsTests +{ + private static readonly DateTimeOffset ConnectedAt = new(2026, 9, 12, 12, 34, 56, TimeSpan.Zero); + private ConnectionStatsResp _response; + + [SetUp] + public async Task SetUp() + { + var provider = new StubConnectionStatsProvider([ + new ConnectionStatsSnapshot( + "connection-1", + "127.0.0.1:50123", + "127.0.0.1:1112", + "projection-catchup", + "gRPC", + "HTTP/2", + true, + ConnectedAt, + 123, + 456, + 7, + 8) + ]); + var serviceType = typeof(Message).Assembly.GetType( + "EventStore.Core.Services.Transport.Grpc.Monitoring", + throwOnError: true); + var service = Activator.CreateInstance( + serviceType!, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + binder: null, + args: [new NoOpPublisher(), provider, new AllowMonitoringAuthorizationProvider()], + culture: null); + + var task = (Task)serviceType!.GetMethod( + nameof(EventStore.Client.Monitoring.Monitoring.MonitoringBase.ConnectionStats))! + .Invoke(service, [new ConnectionStatsReq(), TestServerCallContext.Instance])!; + _response = await task; + } + + [Test] + public void should_map_the_active_connection() + { + var connection = _response.Connections[0]; + + Assert.Multiple(() => + { + Assert.That(connection.ConnectionId, Is.EqualTo("connection-1")); + Assert.That(connection.RemoteEndpoint, Is.EqualTo("127.0.0.1:50123")); + Assert.That(connection.LocalEndpoint, Is.EqualTo("127.0.0.1:1112")); + Assert.That(connection.ClientConnectionName, Is.EqualTo("projection-catchup")); + Assert.That(connection.Application, Is.EqualTo("gRPC")); + Assert.That(connection.Protocol, Is.EqualTo("HTTP/2")); + Assert.That(connection.IsTls, Is.True); + Assert.That(connection.ConnectedAt.ToDateTimeOffset(), Is.EqualTo(ConnectedAt)); + Assert.That(connection.TotalBytesSent, Is.EqualTo(123)); + Assert.That(connection.TotalBytesReceived, Is.EqualTo(456)); + Assert.That(connection.PendingSendBytes, Is.EqualTo(7)); + Assert.That(connection.PendingReceivedBytes, Is.EqualTo(8)); + }); + } + + private sealed class StubConnectionStatsProvider(IReadOnlyList connections) + : IConnectionStatsProvider + { + public IReadOnlyList Snapshot() => connections; + } + + private sealed class NoOpPublisher : IPublisher + { + public void Publish(Message message) + { + } + } + + private sealed class TestServerCallContext : ServerCallContext + { + public static readonly TestServerCallContext Instance = new(); + + private TestServerCallContext() + { + UserStateCore["__HttpContext"] = new DefaultHttpContext(); + } + + protected override string MethodCore => + nameof(EventStore.Client.Monitoring.Monitoring.MonitoringBase.ConnectionStats); + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:0"; + protected override DateTime DeadlineCore => DateTime.MaxValue; + protected override Metadata RequestHeadersCore { get; } = new(); + protected override CancellationToken CancellationTokenCore => CancellationToken.None; + protected override Metadata ResponseTrailersCore { get; } = new(); + protected override Status StatusCore { get; set; } + protected override WriteOptions WriteOptionsCore { get; set; } + protected override AuthContext AuthContextCore => new(null, new Dictionary>()); + protected override IDictionary UserStateCore { get; } = new Dictionary(); + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions options) => + throw new NotSupportedException(); + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + } +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/MonitoringAuthorizationKestrelTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/MonitoringAuthorizationKestrelTests.cs new file mode 100644 index 0000000000..dac02da130 --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/MonitoringAuthorizationKestrelTests.cs @@ -0,0 +1,208 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Reflection; +using System.Security.Claims; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Client.Monitoring; +using EventStore.Core.Bus; +using EventStore.Core.Messages; +using EventStore.Core.Messaging; +using EventStore.Core.Services.Transport.Grpc; +using EventStore.Plugins.Authorization; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; + +[TestFixture] +public class MonitoringAuthorizationKestrelTests +{ + [Test] + public async Task authorized_caller_can_read_each_monitoring_endpoint() + { + var publisher = new ServingPublisher(); + var connections = new CountingConnections(); + var authorization = new CapturingAuthorizationProvider(true); + using var host = CreateHost(publisher, connections, authorization); + await host.StartAsync(); + var address = host.Services.GetRequiredService() + .Features.Get()!.Addresses.Single(); + using var handler = new SocketsHttpHandler(); + using var httpClient = new HttpClient(handler); + using var channel = GrpcChannel.ForAddress(address, new GrpcChannelOptions { HttpClient = httpClient }); + var client = new EventStore.Client.Monitoring.Monitoring.MonitoringClient(channel); + + var connectionResponse = await client.ConnectionStatsAsync(new ConnectionStatsReq()); + var replicationResponse = await client.ReplicationStatsAsync(new ReplicationStatsReq()); + using var statsCall = client.Stats(new StatsReq { RefreshTimePeriodInMs = 1000 }); + var hasStats = await statsCall.ResponseStream.MoveNext(); + + Assert.Multiple(() => + { + Assert.That(connectionResponse, Is.Not.Null); + Assert.That(replicationResponse, Is.Not.Null); + Assert.That(hasStats, Is.True); + Assert.That(connections.SnapshotCalls, Is.EqualTo(1)); + Assert.That(publisher.PublishCalls, Is.EqualTo(2)); + Assert.That(authorization.Operations, Is.EquivalentTo(new[] { + new Operation(Operations.Node.Statistics.Read), + new Operation(Operations.Node.Statistics.Replication), + new Operation(Operations.Node.Statistics.Read) + })); + }); + } + + [Test] + public async Task denied_caller_cannot_read_any_monitoring_data() + { + var publisher = new CountingPublisher(); + var connections = new CountingConnections(); + var authorization = new CapturingAuthorizationProvider(false); + using var host = CreateHost(publisher, connections, authorization); + await host.StartAsync(); + var address = host.Services.GetRequiredService() + .Features.Get()!.Addresses.Single(); + using var handler = new SocketsHttpHandler(); + using var httpClient = new HttpClient(handler); + using var channel = GrpcChannel.ForAddress(address, new GrpcChannelOptions { HttpClient = httpClient }); + var client = new EventStore.Client.Monitoring.Monitoring.MonitoringClient(channel); + + var connectionError = Assert.ThrowsAsync(async () => + await client.ConnectionStatsAsync(new ConnectionStatsReq())); + var replicationError = Assert.ThrowsAsync(async () => + await client.ReplicationStatsAsync(new ReplicationStatsReq())); + var statsError = Assert.ThrowsAsync(async () => + { + using var call = client.Stats(new StatsReq { RefreshTimePeriodInMs = 1000 }); + await call.ResponseStream.MoveNext(); + }); + + Assert.Multiple(() => + { + Assert.That(connectionError!.StatusCode, Is.EqualTo(StatusCode.PermissionDenied)); + Assert.That(replicationError!.StatusCode, Is.EqualTo(StatusCode.PermissionDenied)); + Assert.That(statsError!.StatusCode, Is.EqualTo(StatusCode.PermissionDenied)); + Assert.That(connections.SnapshotCalls, Is.Zero); + Assert.That(publisher.PublishCalls, Is.Zero); + Assert.That(authorization.Operations, Is.EquivalentTo(new[] { + new Operation(Operations.Node.Statistics.Read), + new Operation(Operations.Node.Statistics.Replication), + new Operation(Operations.Node.Statistics.Read) + })); + }); + } + + private static IHost CreateHost(IPublisher publisher, IConnectionStatsProvider connections, + IAuthorizationProvider authorization) => + new HostBuilder() + .ConfigureWebHost(webHost => webHost + .UseKestrel(server => server.Listen(IPAddress.Loopback, 0, + listenOptions => listenOptions.Protocols = HttpProtocols.Http2)) + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(publisher); + services.AddSingleton(connections); + services.AddSingleton(authorization); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(endpoints => endpoints.MapGrpcService()); + })) + .Build(); + + public sealed class ProductionMonitoringAdapter : EventStore.Client.Monitoring.Monitoring.MonitoringBase + { + private readonly object _service; + private readonly Type _serviceType; + + public ProductionMonitoringAdapter(IPublisher publisher, IConnectionStatsProvider connections, + IAuthorizationProvider authorization) + { + _serviceType = typeof(IConnectionStatsProvider).Assembly.GetType( + "EventStore.Core.Services.Transport.Grpc.Monitoring", throwOnError: true)!; + _service = Activator.CreateInstance(_serviceType, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + binder: null, args: [publisher, connections, authorization], culture: null)!; + } + + public override Task ConnectionStats(ConnectionStatsReq request, ServerCallContext context) => + Invoke>(nameof(ConnectionStats), request, context); + + public override Task ReplicationStats(ReplicationStatsReq request, ServerCallContext context) => + Invoke>(nameof(ReplicationStats), request, context); + + public override Task Stats(StatsReq request, IServerStreamWriter responseStream, + ServerCallContext context) => + Invoke(nameof(Stats), request, responseStream, context); + + private T Invoke(string name, params object[] args) => + (T)_serviceType.GetMethod(name)!.Invoke(_service, args)!; + } + + private sealed class CountingPublisher : IPublisher + { + public int PublishCalls { get; private set; } + public void Publish(Message message) + { + PublishCalls++; + throw new InvalidOperationException($"Unexpected message {message.GetType().Name}"); + } + } + + private sealed class ServingPublisher : IPublisher + { + public int PublishCalls { get; private set; } + public void Publish(Message message) + { + PublishCalls++; + switch (message) + { + case MonitoringMessage.GetFreshStats request: + request.Envelope.ReplyWith(new MonitoringMessage.GetFreshStatsCompleted( + true, new Dictionary { ["test"] = 1 })); + break; + case ReplicationMessage.GetReplicationStats request: + request.Envelope.ReplyWith(new ReplicationMessage.GetReplicationStatsCompleted([])); + break; + default: + throw new InvalidOperationException($"Unexpected message {message.GetType().Name}"); + } + } + } + + private sealed class CountingConnections : IConnectionStatsProvider + { + public int SnapshotCalls { get; private set; } + public IReadOnlyList Snapshot() + { + SnapshotCalls++; + return Array.Empty(); + } + } + + private sealed class CapturingAuthorizationProvider(bool allow) : AuthorizationProviderBase + { + public List Operations { get; } = new(); + + public override ValueTask CheckAccessAsync(ClaimsPrincipal principal, Operation operation, + CancellationToken cancellationToken) + { + Operations.Add(operation); + return ValueTask.FromResult(allow); + } + } +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ReplicationStatsTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ReplicationStatsTests.cs index beb71ad0a9..00d8794821 100644 --- a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ReplicationStatsTests.cs +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/ReplicationStatsTests.cs @@ -8,6 +8,7 @@ using EventStore.Core.Messages; using EventStore.Core.Messaging; using Grpc.Core; +using Microsoft.AspNetCore.Http; using NUnit.Framework; using CoreReplicationStats = EventStore.Core.Messages.ReplicationMessage.ReplicationStats; @@ -51,7 +52,7 @@ public async Task SetUp() serviceType!, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, binder: null, - args: [_publisher], + args: [_publisher, null, new AllowMonitoringAuthorizationProvider()], culture: null); var task = (Task)serviceType!.GetMethod( @@ -115,6 +116,7 @@ private sealed class TestServerCallContext : ServerCallContext private TestServerCallContext() { + UserStateCore["__HttpContext"] = new DefaultHttpContext(); } protected override string MethodCore => diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/StatsRpcTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/StatsRpcTests.cs index 953b64ebbe..99e8e4bed5 100644 --- a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/StatsRpcTests.cs +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/StatsRpcTests.cs @@ -10,6 +10,7 @@ using EventStore.Core.Messaging; using EventStore.Core.Services.Monitoring.Stats; using Grpc.Core; +using Microsoft.AspNetCore.Http; using NUnit.Framework; namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; @@ -149,7 +150,7 @@ private static async Task ReadSingleResponse(CapturingPublisher publi serviceType!, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, binder: null, - args: [publisher], + args: [publisher, null, new AllowMonitoringAuthorizationProvider()], culture: null); using var cts = new CancellationTokenSource(); @@ -217,14 +218,21 @@ public Task WriteAsync(StatsResp message) } } - private sealed class TestServerCallContext(CancellationToken cancellationToken) : ServerCallContext + private sealed class TestServerCallContext : ServerCallContext { + private readonly CancellationToken _cancellationToken; + + public TestServerCallContext(CancellationToken cancellationToken) + { + _cancellationToken = cancellationToken; + UserStateCore["__HttpContext"] = new DefaultHttpContext(); + } protected override string MethodCore => nameof(EventStore.Client.Monitoring.Monitoring.MonitoringBase.Stats); protected override string HostCore => "localhost"; protected override string PeerCore => "ipv4:127.0.0.1:0"; protected override DateTime DeadlineCore => DateTime.MaxValue; protected override Metadata RequestHeadersCore { get; } = new(); - protected override CancellationToken CancellationTokenCore => cancellationToken; + protected override CancellationToken CancellationTokenCore => _cancellationToken; protected override Metadata ResponseTrailersCore { get; } = new(); protected override Status StatusCore { get; set; } protected override WriteOptions WriteOptionsCore { get; set; } diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/TcpStatsTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/TcpStatsTests.cs deleted file mode 100644 index f46a618245..0000000000 --- a/src/EventStore.Core.Tests/Services/Transport/Grpc/MonitoringTests/TcpStatsTests.cs +++ /dev/null @@ -1,136 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Reflection; -using System.Threading; -using System.Threading.Tasks; -using EventStore.Client.Monitoring; -using EventStore.Core.Bus; -using EventStore.Core.Messages; -using EventStore.Core.Messaging; -using Grpc.Core; -using NUnit.Framework; -using CoreTcpConnectionStats = EventStore.Core.Messages.MonitoringMessage.TcpConnectionStats; - -namespace EventStore.Core.Tests.Services.Transport.Grpc.MonitoringTests; - -[TestFixture] -public class TcpStatsTests -{ - private readonly Guid _connectionId = Guid.Parse("1e1c6d68-3c7c-446f-915e-8bdf8d35e122"); - private TcpStatsResp _response; - private CapturingPublisher _publisher; - - [SetUp] - public async Task SetUp() - { - _publisher = new CapturingPublisher(new List { - new() { - RemoteEndPoint = "127.0.0.1:1113", - LocalEndPoint = "127.0.0.1:2113", - ClientConnectionName = "test-connection", - ConnectionId = _connectionId, - TotalBytesSent = 123, - TotalBytesReceived = 456, - PendingSendBytes = 7, - PendingReceivedBytes = 8, - IsExternalConnection = true, - IsSslConnection = true - }, - new() { - ConnectionId = Guid.Empty - } - }); - var serviceType = typeof(MonitoringMessage).Assembly.GetType( - "EventStore.Core.Services.Transport.Grpc.Monitoring", - throwOnError: true); - var service = Activator.CreateInstance( - serviceType!, - BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, - binder: null, - args: [_publisher], - culture: null); - - var task = (Task)serviceType!.GetMethod(nameof(EventStore.Client.Monitoring.Monitoring.MonitoringBase.TcpStats))! - .Invoke(service, [new TcpStatsReq(), TestServerCallContext.Instance])!; - _response = await task; - } - - [Test] - public void should_request_fresh_tcp_connection_stats() - { - Assert.IsTrue(_publisher.RequestedTcpStats); - } - - [Test] - public void should_return_the_tcp_connection_stats() - { - Assert.AreEqual(2, _response.Connections.Count); - } - - [Test] - public void should_map_all_tcp_connection_fields() - { - var connection = _response.Connections[0]; - - Assert.AreEqual("127.0.0.1:1113", connection.RemoteEndpoint); - Assert.AreEqual("127.0.0.1:2113", connection.LocalEndpoint); - Assert.AreEqual("test-connection", connection.ClientConnectionName); - Assert.AreEqual(_connectionId.ToString("D"), connection.ConnectionId); - Assert.AreEqual(123, connection.TotalBytesSent); - Assert.AreEqual(456, connection.TotalBytesReceived); - Assert.AreEqual(7, connection.PendingSendBytes); - Assert.AreEqual(8, connection.PendingReceivedBytes); - Assert.IsTrue(connection.IsExternalConnection); - Assert.IsTrue(connection.IsSslConnection); - } - - [Test] - public void should_map_null_strings_to_empty_values() - { - var connection = _response.Connections[1]; - - Assert.AreEqual(string.Empty, connection.RemoteEndpoint); - Assert.AreEqual(string.Empty, connection.LocalEndpoint); - Assert.AreEqual(string.Empty, connection.ClientConnectionName); - } - - private sealed class CapturingPublisher(List connectionStats) : IPublisher - { - public bool RequestedTcpStats { get; private set; } - - public void Publish(Message message) - { - if (message is not MonitoringMessage.GetFreshTcpConnectionStats request) - { - throw new InvalidOperationException($"Unexpected message {message.GetType().Name}"); - } - - RequestedTcpStats = true; - request.Envelope.ReplyWith(new MonitoringMessage.GetFreshTcpConnectionStatsCompleted(connectionStats)); - } - } - - private sealed class TestServerCallContext : ServerCallContext - { - public static readonly TestServerCallContext Instance = new(); - - private TestServerCallContext() - { - } - - protected override string MethodCore => nameof(EventStore.Client.Monitoring.Monitoring.MonitoringBase.TcpStats); - protected override string HostCore => "localhost"; - protected override string PeerCore => "ipv4:127.0.0.1:0"; - protected override DateTime DeadlineCore => DateTime.MaxValue; - protected override Metadata RequestHeadersCore { get; } = new(); - protected override CancellationToken CancellationTokenCore => CancellationToken.None; - protected override Metadata ResponseTrailersCore { get; } = new(); - protected override Status StatusCore { get; set; } - protected override WriteOptions WriteOptionsCore { get; set; } - protected override AuthContext AuthContextCore => new(null, new Dictionary>()); - protected override IDictionary UserStateCore { get; } = new Dictionary(); - protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions options) => - throw new NotSupportedException(); - protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; - } -} diff --git a/src/EventStore.Core/ClusterVNodeStartup.cs b/src/EventStore.Core/ClusterVNodeStartup.cs index d3767fe04d..fd01cb91ba 100644 --- a/src/EventStore.Core/ClusterVNodeStartup.cs +++ b/src/EventStore.Core/ClusterVNodeStartup.cs @@ -235,7 +235,10 @@ public void ConfigureServicesOnly(IServiceCollection services) .AddSingleton(new Elections(_mainQueue, _authorizationProvider, _clusterDns)) .AddSingleton(new ClientGossip(_mainQueue, _authorizationProvider, _trackers.GossipTrackers.ProcessingRequestFromGrpcClient)) - .AddSingleton(new Monitoring(_monitoringQueue)) + .AddSingleton(serviceProvider => new Monitoring( + _monitoringQueue, + serviceProvider.GetService(), + _authorizationProvider)) .AddSingleton(_nodeInformationProvider) .AddSingleton(new NodeInformation(_nodeInformationProvider, _authorizationProvider)) .AddSingleton(new Redaction(_mainQueue, _authorizationProvider)) diff --git a/src/EventStore.Core/Messages/MonitoringMessage.cs b/src/EventStore.Core/Messages/MonitoringMessage.cs index bde535e546..5616e573cf 100644 --- a/src/EventStore.Core/Messages/MonitoringMessage.cs +++ b/src/EventStore.Core/Messages/MonitoringMessage.cs @@ -205,44 +205,6 @@ public GetFreshStatsCompleted(bool success, Dictionary stats) } } - [DerivedMessage(CoreMessage.Monitoring)] - public partial class GetFreshTcpConnectionStats : Message - { - public readonly IEnvelope Envelope; - - public GetFreshTcpConnectionStats(IEnvelope envelope) - { - Ensure.NotNull(envelope, "envelope"); - - Envelope = envelope; - } - } - - [DerivedMessage(CoreMessage.Monitoring)] - public partial class GetFreshTcpConnectionStatsCompleted : Message - { - public readonly List ConnectionStats; - - public GetFreshTcpConnectionStatsCompleted(List connectionStats) - { - ConnectionStats = connectionStats; - } - } - - public class TcpConnectionStats - { - public string RemoteEndPoint { get; set; } - public string LocalEndPoint { get; set; } - public string ClientConnectionName { get; set; } - public Guid ConnectionId { get; set; } - public long TotalBytesSent { get; set; } - public long TotalBytesReceived { get; set; } - public int PendingSendBytes { get; set; } - public int PendingReceivedBytes { get; set; } - public bool IsExternalConnection { get; set; } - public bool IsSslConnection { get; set; } - } - [DerivedMessage(CoreMessage.Monitoring)] public partial class InternalStatsRequest : Message { diff --git a/src/EventStore.Core/Services/Transport/Grpc/IConnectionStatsProvider.cs b/src/EventStore.Core/Services/Transport/Grpc/IConnectionStatsProvider.cs new file mode 100644 index 0000000000..414f2f70f4 --- /dev/null +++ b/src/EventStore.Core/Services/Transport/Grpc/IConnectionStatsProvider.cs @@ -0,0 +1,30 @@ +using System; +using System.Collections.Generic; + +namespace EventStore.Core.Services.Transport.Grpc; + +public interface IConnectionStatsProvider +{ + IReadOnlyList Snapshot(); +} + +public record ConnectionStatsSnapshot( + string ConnectionId, + string RemoteEndPoint, + string LocalEndPoint, + string ClientName, + string Application, + string Protocol, + bool IsTls, + DateTimeOffset ConnectedAt, + long TotalBytesSent, + long TotalBytesReceived, + long PendingSendBytes, + long PendingReceivedBytes); + +internal sealed class EmptyConnectionStatsProvider : IConnectionStatsProvider +{ + public static readonly EmptyConnectionStatsProvider Instance = new(); + + public IReadOnlyList Snapshot() => Array.Empty(); +} diff --git a/src/EventStore.Core/Services/Transport/Grpc/Monitoring.cs b/src/EventStore.Core/Services/Transport/Grpc/Monitoring.cs index 1518bf82a3..7fb07df1b6 100644 --- a/src/EventStore.Core/Services/Transport/Grpc/Monitoring.cs +++ b/src/EventStore.Core/Services/Transport/Grpc/Monitoring.cs @@ -4,6 +4,8 @@ using EventStore.Core.Bus; using EventStore.Core.Messages; using EventStore.Core.Messaging; +using EventStore.Plugins.Authorization; +using Google.Protobuf.WellKnownTypes; using Grpc.Core; namespace EventStore.Core.Services.Transport.Grpc @@ -11,9 +13,14 @@ namespace EventStore.Core.Services.Transport.Grpc internal partial class Monitoring : EventStore.Client.Monitoring.Monitoring.MonitoringBase { private readonly IPublisher _publisher; + private readonly IConnectionStatsProvider _connectionStatsProvider; + private readonly IAuthorizationProvider _authorizationProvider; + private static readonly Operation ReadStatisticsOperation = new(Plugins.Authorization.Operations.Node.Statistics.Read); + private static readonly Operation ReadReplicationStatisticsOperation = new(Plugins.Authorization.Operations.Node.Statistics.Replication); - public override Task Stats(StatsReq request, IServerStreamWriter responseStream, ServerCallContext context) + public override async Task Stats(StatsReq request, IServerStreamWriter responseStream, ServerCallContext context) { + await RequireAccess(ReadStatisticsOperation, context); var useGrouping = request.HasUseGrouping ? request.UseGrouping : false; if (!useGrouping && !string.IsNullOrEmpty(request.StatsPath)) { @@ -22,7 +29,7 @@ public override Task Stats(StatsReq request, IServerStreamWriter resp "Dynamic stats selection works only with grouping enabled")); } - return StreamStats(); + await StreamStats(); async Task StreamStats() { @@ -84,45 +91,37 @@ async Task StreamStats() } } - public override Task TcpStats(TcpStatsReq request, ServerCallContext context) + public override async Task ConnectionStats( + ConnectionStatsReq request, + ServerCallContext context) { - var responseSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var envelope = new CallbackEnvelope(message => + await RequireAccess(ReadStatisticsOperation, context); + var response = new ConnectionStatsResp(); + foreach (var connection in _connectionStatsProvider.Snapshot()) { - if (message is not MonitoringMessage.GetFreshTcpConnectionStatsCompleted completed) - { - responseSource.TrySetException( - UnknownMessage(message)); - return; - } - - var response = new TcpStatsResp(); - foreach (var connection in completed.ConnectionStats) + response.Connections.Add(new EventStore.Client.Monitoring.ConnectionStats { - response.Connections.Add(new TcpConnectionStats - { - RemoteEndpoint = connection.RemoteEndPoint ?? string.Empty, - LocalEndpoint = connection.LocalEndPoint ?? string.Empty, - ClientConnectionName = connection.ClientConnectionName ?? string.Empty, - ConnectionId = connection.ConnectionId.ToString("D"), - TotalBytesSent = connection.TotalBytesSent, - TotalBytesReceived = connection.TotalBytesReceived, - PendingSendBytes = connection.PendingSendBytes, - PendingReceivedBytes = connection.PendingReceivedBytes, - IsExternalConnection = connection.IsExternalConnection, - IsSslConnection = connection.IsSslConnection - }); - } - - responseSource.TrySetResult(response); - }); + RemoteEndpoint = connection.RemoteEndPoint ?? string.Empty, + LocalEndpoint = connection.LocalEndPoint ?? string.Empty, + ClientConnectionName = connection.ClientName ?? string.Empty, + ConnectionId = connection.ConnectionId ?? string.Empty, + TotalBytesSent = connection.TotalBytesSent, + TotalBytesReceived = connection.TotalBytesReceived, + PendingSendBytes = connection.PendingSendBytes, + PendingReceivedBytes = connection.PendingReceivedBytes, + IsTls = connection.IsTls, + Protocol = connection.Protocol ?? string.Empty, + Application = connection.Application ?? string.Empty, + ConnectedAt = Timestamp.FromDateTimeOffset(connection.ConnectedAt) + }); + } - _publisher.Publish(new MonitoringMessage.GetFreshTcpConnectionStats(envelope)); - return responseSource.Task.WaitAsync(context.CancellationToken); + return response; } - public override Task ReplicationStats(ReplicationStatsReq request, ServerCallContext context) + public override async Task ReplicationStats(ReplicationStatsReq request, ServerCallContext context) { + await RequireAccess(ReadReplicationStatisticsOperation, context); var responseSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var envelope = new CallbackEnvelope(message => @@ -154,12 +153,24 @@ public override Task ReplicationStats(ReplicationStatsReq }); _publisher.Publish(new ReplicationMessage.GetReplicationStats(envelope)); - return responseSource.Task.WaitAsync(context.CancellationToken); + return await responseSource.Task.WaitAsync(context.CancellationToken); } - public Monitoring(IPublisher publisher) + public Monitoring(IPublisher publisher, IConnectionStatsProvider connectionStatsProvider, + IAuthorizationProvider authorizationProvider) { _publisher = publisher; + _connectionStatsProvider = connectionStatsProvider ?? EmptyConnectionStatsProvider.Instance; + _authorizationProvider = authorizationProvider ?? throw new ArgumentNullException(nameof(authorizationProvider)); + } + + private async Task RequireAccess(Operation operation, ServerCallContext context) + { + if (!await _authorizationProvider.CheckAccessAsync( + context.GetHttpContext().User, operation, context.CancellationToken)) + { + throw RpcExceptions.AccessDenied(); + } } private static Exception UnknownMessage(Message message) where T : Message => diff --git a/src/Protos/Grpc/monitoring.proto b/src/Protos/Grpc/monitoring.proto index dc1a31db9e..76ef60d2df 100644 --- a/src/Protos/Grpc/monitoring.proto +++ b/src/Protos/Grpc/monitoring.proto @@ -3,10 +3,11 @@ package event_store.client.monitoring; option java_package = "com.eventstore.dbclient.proto.monitoring"; import "google/protobuf/struct.proto"; +import "google/protobuf/timestamp.proto"; service Monitoring { rpc Stats(StatsReq) returns (stream StatsResp); - rpc TcpStats(TcpStatsReq) returns (TcpStatsResp); + rpc ConnectionStats(ConnectionStatsReq) returns (ConnectionStatsResp); rpc ReplicationStats(ReplicationStatsReq) returns (ReplicationStatsResp); } @@ -22,24 +23,26 @@ message StatsResp { google.protobuf.Value structured_stats = 2; } -message TcpStatsReq { +message ConnectionStatsReq { } -message TcpStatsResp { - repeated TcpConnectionStats connections = 1; +message ConnectionStatsResp { + repeated ConnectionStats connections = 1; } -message TcpConnectionStats { +message ConnectionStats { string remote_endpoint = 1; string local_endpoint = 2; string client_connection_name = 3; string connection_id = 4; int64 total_bytes_sent = 5; int64 total_bytes_received = 6; - int32 pending_send_bytes = 7; - int32 pending_received_bytes = 8; - bool is_external_connection = 9; - bool is_ssl_connection = 10; + int64 pending_send_bytes = 7; + int64 pending_received_bytes = 8; + bool is_tls = 9; + string protocol = 10; + string application = 11; + google.protobuf.Timestamp connected_at = 12; } message ReplicationStatsReq {