Skip to content

Commit 3143c08

Browse files
yordisyordis
authored
fix(cluster): preserve internal transport isolation (#495)
- Client-facing traffic must not share the listener used for replication, gossip, elections, and request forwarding. - Explicit listener roles and transport capabilities keep future dedicated endpoints from inheriting client behavior by accident. - Existing configuration names and the established membership wire field remain stable to avoid unnecessary operator and protocol churn. - Strict endpoint isolation requires a coordinated cluster restart because older nodes do not advertise a separate internal cluster endpoint. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
1 parent 62a862f commit 3143c08

39 files changed

Lines changed: 874 additions & 241 deletions

File tree

‎docker-compose.yml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ services:
1616
env_file:
1717
- shared.env
1818
environment:
19-
- EVENTSTORE_GOSSIP_SEED=172.30.240.12:2113,172.30.240.13:2113
19+
- EVENTSTORE_GOSSIP_SEED=172.30.240.12:1112,172.30.240.13:1112
2020
- EVENTSTORE_REPLICATION_IP=172.30.240.11
2121
- EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node/node.crt
2222
- EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node/node.key
@@ -43,7 +43,7 @@ services:
4343
env_file:
4444
- shared.env
4545
environment:
46-
- EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.13:2113
46+
- EVENTSTORE_GOSSIP_SEED=172.30.240.11:1112,172.30.240.13:1112
4747
- EVENTSTORE_REPLICATION_IP=172.30.240.12
4848
- EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node/node.crt
4949
- EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node/node.key
@@ -70,7 +70,7 @@ services:
7070
env_file:
7171
- shared.env
7272
environment:
73-
- EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.12:2113
73+
- EVENTSTORE_GOSSIP_SEED=172.30.240.11:1112,172.30.240.12:1112
7474
- EVENTSTORE_REPLICATION_IP=172.30.240.13
7575
- EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node/node.crt
7676
- EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node/node.key

‎proto.lock‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8124,4 +8124,4 @@
81248124
}
81258125
}
81268126
]
8127-
}
8127+
}

‎src/EventStore.ClusterNode/Components/Services/ClusterStatusService.cs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,7 +211,9 @@ private static ClientClusterInfo.ClientMemberInfo FindMemberByInternalEndpoint(
211211
string endpoint)
212212
{
213213
var cleaned = endpoint.Replace("Unspecified/", "", StringComparison.OrdinalIgnoreCase);
214-
return members.FirstOrDefault(x => string.Equals(InternalTcpEndpoint(x), cleaned, StringComparison.OrdinalIgnoreCase));
214+
return members.FirstOrDefault(x =>
215+
string.Equals(ReplicationEndpoint(x), cleaned, StringComparison.OrdinalIgnoreCase) ||
216+
string.Equals(InternalTcpEndpoint(x), cleaned, StringComparison.OrdinalIgnoreCase));
215217
}
216218

217219
private static Uri BuildLeaderAddress(
@@ -224,6 +226,9 @@ private static string InternalTcpEndpoint(ClientClusterInfo.ClientMemberInfo mem
224226
member.InternalTcpIp,
225227
member.InternalSecureTcpPort != 0 ? member.InternalSecureTcpPort : member.InternalTcpPort);
226228

229+
private static string ReplicationEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
230+
Endpoint(member.ClusterEndPointIp, member.ClusterEndPointPort);
231+
227232
private static string HttpEndpoint(ClientClusterInfo.ClientMemberInfo member) =>
228233
Endpoint(member.HttpEndPointIp, member.HttpEndPointPort);
229234

‎src/EventStore.ClusterNode/Components/Services/EndpointPolicy.cs‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Net;
4+
using Google.Protobuf.Reflection;
5+
using Microsoft.AspNetCore.Http;
6+
using Microsoft.AspNetCore.Server.Kestrel.Core;
7+
8+
namespace EventStore.ClusterNode.Components.Services;
9+
10+
public enum EndpointRole
11+
{
12+
Client,
13+
Cluster,
14+
}
15+
16+
public sealed record EndpointBinding(
17+
EndpointRole Role,
18+
IPEndPoint ListenEndPoint,
19+
HttpProtocols Protocols);
20+
21+
public sealed record GrpcEndpointRoute(ServiceDescriptor Service, EndpointRole Role)
22+
{
23+
public PathString Path => new($"/{Service.FullName}");
24+
}
25+
26+
public sealed class EndpointPolicy
27+
{
28+
private readonly IReadOnlyList<EndpointBinding> _bindings;
29+
private readonly IReadOnlyList<GrpcEndpointRoute> _routes;
30+
private readonly EndpointRole _defaultRouteRole;
31+
private readonly EndpointRole _nonIpEndpointRole;
32+
33+
public EndpointPolicy(
34+
IReadOnlyList<EndpointBinding> bindings,
35+
IReadOnlyList<GrpcEndpointRoute> routes,
36+
EndpointRole defaultRouteRole,
37+
EndpointRole nonIpEndpointRole)
38+
{
39+
_bindings = bindings;
40+
_routes = routes;
41+
_defaultRouteRole = defaultRouteRole;
42+
_nonIpEndpointRole = nonIpEndpointRole;
43+
}
44+
45+
public bool Allows(HttpContext context)
46+
{
47+
var endpointRole = GetEndpointRole(context.Connection.LocalIpAddress, context.Connection.LocalPort);
48+
return endpointRole.HasValue && endpointRole.Value == GetRouteRole(context.Request.Path);
49+
}
50+
51+
private EndpointRole GetRouteRole(PathString requestPath)
52+
{
53+
foreach (var route in _routes)
54+
{
55+
if (requestPath.StartsWithSegments(route.Path, StringComparison.Ordinal))
56+
return route.Role;
57+
}
58+
59+
return _defaultRouteRole;
60+
}
61+
62+
private EndpointRole? GetEndpointRole(IPAddress localAddress, int localPort)
63+
{
64+
if (localAddress is null)
65+
return _nonIpEndpointRole;
66+
67+
foreach (var binding in _bindings)
68+
{
69+
if (!Matches(binding.ListenEndPoint, localAddress, localPort))
70+
continue;
71+
72+
return binding.Role;
73+
}
74+
75+
return null;
76+
}
77+
78+
private static bool Matches(IPEndPoint listenEndPoint, IPAddress localAddress, int localPort)
79+
{
80+
if (localPort != listenEndPoint.Port)
81+
return false;
82+
83+
return listenEndPoint.Address.Equals(IPAddress.Any) ||
84+
listenEndPoint.Address.Equals(IPAddress.IPv6Any) ||
85+
listenEndPoint.Address.Equals(localAddress);
86+
}
87+
}

‎src/EventStore.ClusterNode/Program.cs‎

Lines changed: 45 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
using Microsoft.AspNetCore.Builder;
2626
using Microsoft.AspNetCore.DataProtection;
2727
using Microsoft.AspNetCore.Hosting;
28+
using Microsoft.AspNetCore.Http;
2829
using Microsoft.AspNetCore.Server.Kestrel.Core;
2930
using Microsoft.Extensions.Configuration;
3031
using Microsoft.Extensions.DependencyInjection;
@@ -272,6 +273,25 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
272273
{
273274
x.SuppressStatusMessages = true;
274275
});
276+
EndpointBinding[] endpointBindings =
277+
[
278+
new(EndpointRole.Client,
279+
new System.Net.IPEndPoint(options.Interface.NodeIp, options.Interface.NodePort),
280+
HttpProtocols.Http1AndHttp2),
281+
new(EndpointRole.Cluster,
282+
options.Interface.GetClusterListenEndPoint(),
283+
HttpProtocols.Http2),
284+
];
285+
var endpointPolicy = new EndpointPolicy(
286+
endpointBindings,
287+
[
288+
new(EventStore.Cluster.Gossip.Descriptor, EndpointRole.Cluster),
289+
new(EventStore.Cluster.Elections.Descriptor, EndpointRole.Cluster),
290+
new(EventStore.Replication.Replication.Descriptor, EndpointRole.Cluster),
291+
new(EventStore.Forwarding.RequestForwarding.Descriptor, EndpointRole.Cluster),
292+
],
293+
defaultRouteRole: EndpointRole.Client,
294+
nonIpEndpointRole: EndpointRole.Client);
275295

276296
builder.WebHost.ConfigureKestrel(server =>
277297
{
@@ -280,9 +300,13 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
280300
server.Limits.Http2.KeepAlivePingTimeout =
281301
TimeSpan.FromMilliseconds(options.Grpc.KeepAliveTimeout);
282302

283-
server.Listen(options.Interface.NodeIp, options.Interface.NodePort, listenOptions =>
284-
ConfigureHttpOptions(listenOptions, hostedService,
285-
useHttps: !hostedService.Node.DisableHttps));
303+
foreach (var binding in endpointBindings)
304+
{
305+
server.Listen(binding.ListenEndPoint, listenOptions =>
306+
ConfigureHttpOptions(listenOptions, hostedService,
307+
useHttps: !hostedService.Node.DisableHttps,
308+
protocols: binding.Protocols));
309+
}
286310

287311
if (hostedService.Node.EnableUnixSocket)
288312
{
@@ -325,6 +349,16 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
325349
builder.Services.AddSingleton<IHostedService>(hostedService);
326350

327351
var app = builder.Build();
352+
app.Use(async (context, next) =>
353+
{
354+
if (!endpointPolicy.Allows(context))
355+
{
356+
context.Response.StatusCode = StatusCodes.Status404NotFound;
357+
return;
358+
}
359+
360+
await next(context);
361+
});
328362
app.UseMiddleware<UiCredentialsMiddleware>();
329363
hostedService.Node.Startup.Configure(app);
330364
if (oauthEnabled)
@@ -376,14 +410,19 @@ async Task Run(ClusterVNodeHostedService hostedService, ManualResetEventSlim sig
376410
}
377411
}
378412

379-
private static void ConfigureHttpOptions(ListenOptions listenOptions, ClusterVNodeHostedService hostedService,
380-
bool useHttps)
413+
private static void ConfigureHttpOptions(
414+
ListenOptions listenOptions,
415+
ClusterVNodeHostedService hostedService,
416+
bool useHttps,
417+
HttpProtocols protocols = HttpProtocols.Http1AndHttp2)
381418
{
419+
listenOptions.Protocols = protocols;
420+
382421
if (useHttps)
383422
{
384423
listenOptions.UseHttps(CreateServerOptionsSelectionCallback(hostedService), null);
385424
}
386-
else
425+
else if (protocols != HttpProtocols.Http2)
387426
{
388427
listenOptions.Use(next =>
389428
new ClearTextHttpMultiplexingMiddleware(next).OnConnectAsync);

‎src/EventStore.Core.Tests/Cluster/MemberInfoTests.cs‎

Lines changed: 37 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -60,12 +60,12 @@ public void member_with_ip_endpoint_should_equal()
6060
}
6161

6262
[Test]
63-
public void grpc_round_trip_preserves_tcp_and_replication_endpoints()
63+
public void grpc_round_trip_preserves_tcp_and_cluster_endpoints()
6464
{
6565
var member = EventStore.Core.Cluster.MemberInfo.Initial(Guid.NewGuid(), DateTime.UtcNow,
6666
VNodeState.Unknown, true,
6767
InternalTcp, null, null, ExternalSecureTcp, Http,
68-
"client", 2113, 1113, 0, false, replicationEndPoint: Replication);
68+
"client", 2113, 1113, 0, false, clusterEndPoint: Replication);
6969

7070
var result = FromGrpcClusterInfo(ToGrpcClusterInfo(
7171
new EventStore.Core.Cluster.ClusterInfo(member))).Members[0];
@@ -75,11 +75,11 @@ public void grpc_round_trip_preserves_tcp_and_replication_endpoints()
7575
Assert.That(result.ExternalTcpEndPoint, Is.Null);
7676
Assert.That(result.ExternalSecureTcpEndPoint, Is.EqualTo(ExternalSecureTcp));
7777
Assert.That(result.HttpEndPoint, Is.EqualTo(Http));
78-
Assert.That(result.ReplicationEndPoint, Is.EqualTo(Replication));
78+
Assert.That(result.ClusterEndPoint, Is.EqualTo(Replication));
7979
}
8080

8181
[Test]
82-
public void explicit_replication_endpoint_is_recognized_without_replacing_tcp_endpoints()
82+
public void explicit_cluster_endpoint_is_recognized_without_replacing_tcp_endpoints()
8383
{
8484
var member = CreateMember(Replication);
8585
var vnode = new VNodeInfo(Guid.NewGuid(), 0,
@@ -95,22 +95,42 @@ public void explicit_replication_endpoint_is_recognized_without_replacing_tcp_en
9595
Assert.That(member.InternalSecureTcpEndPoint, Is.EqualTo(InternalSecureTcp));
9696
Assert.That(member.ExternalTcpEndPoint, Is.EqualTo(ExternalTcp));
9797
Assert.That(member.ExternalSecureTcpEndPoint, Is.EqualTo(ExternalSecureTcp));
98-
Assert.That(vnode.ReplicationEndPoint, Is.SameAs(Replication));
99-
Assert.That(advertise.ReplicationEndPoint, Is.SameAs(Replication));
98+
Assert.That(vnode.ClusterEndPoint, Is.SameAs(Replication));
99+
Assert.That(advertise.ClusterEndPoint, Is.SameAs(Replication));
100100
}
101101

102102
[Test]
103-
public void client_member_preserves_the_replication_endpoint()
103+
public void client_member_preserves_the_cluster_endpoint()
104104
{
105105
var clientMember = new EventStore.Core.Cluster.ClientClusterInfo.ClientMemberInfo(
106106
CreateMember(Replication));
107107

108-
Assert.That(clientMember.ReplicationEndPointIp, Is.EqualTo(Replication.Host));
109-
Assert.That(clientMember.ReplicationEndPointPort, Is.EqualTo(Replication.Port));
108+
Assert.That(clientMember.ClusterEndPointIp, Is.EqualTo(Replication.Host));
109+
Assert.That(clientMember.ClusterEndPointPort, Is.EqualTo(Replication.Port));
110110
}
111111

112112
[Test]
113-
public void missing_replication_endpoint_falls_back_to_http_endpoint()
113+
public void client_cluster_info_excludes_internal_discovery_placeholders()
114+
{
115+
var member = CreateMember(Replication);
116+
var seed = EventStore.Core.Cluster.MemberInfo.ForManager(
117+
Guid.Empty,
118+
DateTime.UtcNow,
119+
true,
120+
Replication,
121+
clusterEndPoint: Replication);
122+
123+
var clientCluster = new EventStore.Core.Cluster.ClientClusterInfo(
124+
new EventStore.Core.Cluster.ClusterInfo(member, seed),
125+
Http.Host,
126+
Http.Port);
127+
128+
Assert.That(clientCluster.Members, Has.Length.EqualTo(1));
129+
Assert.That(clientCluster.Members[0].InstanceId, Is.EqualTo(member.InstanceId));
130+
}
131+
132+
[Test]
133+
public void missing_cluster_endpoint_falls_back_to_http_endpoint()
114134
{
115135
var member = CreateMember();
116136
var vnode = new VNodeInfo(Guid.NewGuid(), 0,
@@ -121,28 +141,28 @@ public void missing_replication_endpoint_falls_back_to_http_endpoint()
121141
InternalTcp, InternalSecureTcp, ExternalTcp, ExternalSecureTcp, Http,
122142
null, null, 0, null, 0, 0);
123143

124-
Assert.That(member.ReplicationEndPoint, Is.SameAs(Http));
125-
Assert.That(vnode.ReplicationEndPoint, Is.SameAs(Http));
126-
Assert.That(advertise.ReplicationEndPoint, Is.SameAs(Http));
144+
Assert.That(member.ClusterEndPoint, Is.SameAs(Http));
145+
Assert.That(vnode.ClusterEndPoint, Is.SameAs(Http));
146+
Assert.That(advertise.ClusterEndPoint, Is.SameAs(Http));
127147
}
128148

129149
[Test]
130-
public void grpc_member_without_replication_endpoint_falls_back_to_http_endpoint()
150+
public void grpc_member_without_cluster_endpoint_falls_back_to_http_endpoint()
131151
{
132152
var grpcCluster = ToGrpcClusterInfo(
133153
new EventStore.Core.Cluster.ClusterInfo(CreateMember(Replication)));
134154
grpcCluster.Members[0].ReplicationEndPoint = null;
135155

136156
var result = FromGrpcClusterInfo(grpcCluster).Members[0];
137157

138-
Assert.That(result.ReplicationEndPoint, Is.EqualTo(Http));
158+
Assert.That(result.ClusterEndPoint, Is.EqualTo(Http));
139159
}
140160

141-
private static EventStore.Core.Cluster.MemberInfo CreateMember(DnsEndPoint replicationEndPoint = null) =>
161+
private static EventStore.Core.Cluster.MemberInfo CreateMember(DnsEndPoint clusterEndPoint = null) =>
142162
EventStore.Core.Cluster.MemberInfo.Initial(Guid.NewGuid(), DateTime.UtcNow,
143163
VNodeState.Unknown, true,
144164
InternalTcp, InternalSecureTcp, ExternalTcp, ExternalSecureTcp, Http,
145-
"client", 2113, 1113, 0, false, replicationEndPoint: replicationEndPoint);
165+
"client", 2113, 1113, 0, false, clusterEndPoint: clusterEndPoint);
146166

147167
private static EventStore.Cluster.ClusterInfo ToGrpcClusterInfo(
148168
EventStore.Core.Cluster.ClusterInfo clusterInfo) =>

0 commit comments

Comments
 (0)