diff --git a/src/StackExchange.Redis/ClusterTopology.cs b/src/StackExchange.Redis/ClusterTopology.cs index a6ae7e3c2..52b549bfa 100644 --- a/src/StackExchange.Redis/ClusterTopology.cs +++ b/src/StackExchange.Redis/ClusterTopology.cs @@ -110,6 +110,12 @@ internal ClusterTopologyNode(ClusterSlotNode node, string? primaryNodeId) /// public IReadOnlyList Identities => _identities ?? []; + /// + /// The announced endpoint form to use for dialing. Companion identities such as an IP address must not be + /// registered as separate dialable servers when this preferred endpoint is a hostname. + /// + public EndPoint? PreferredDialEndPoint => Identities.Count > 0 ? Identities[0] : null; + /// The slots this node serves as primary; empty for a replica. public IReadOnlyList Slots => _slots ?? []; diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 40de8af22..90160da65 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -43,6 +43,7 @@ public sealed partial class ConnectionMultiplexer : IInternalConnectionMultiplex // announced hostname), and `servers` is keyed on exact endpoint equality, so without this the same // node reached by its other name becomes a second ServerEndPoint - see #2826 private readonly ConcurrentDictionary _serverIdentities = new(); + private readonly ConcurrentDictionary _preferredDialEndpoints = new(); private volatile ServerSnapshot _serverSnapshot = ServerSnapshot.Empty; private volatile bool _isDisposed; @@ -1006,6 +1007,9 @@ ServerEndPoint IInternalConnectionMultiplexer.GetServerEndPoint(EndPoint endpoin : null; } + internal EndPoint GetDialEndPoint(ServerEndPoint server) + => _preferredDialEndpoints.TryGetValue(server, out var endpoint) ? endpoint : server.EndPoint; + // bumped once per applied cluster topology; absence is measured in these rather than in time, so a // quiet client cannot age an endpoint out simply by not reconfiguring private int _topologyGeneration; @@ -1033,19 +1037,19 @@ internal async Task ApplyTopologyGenerationAsync(ClusterTopology topology, ILogg List? merge = null; foreach (var node in topology.Nodes) { - ServerEndPoint? survivor = null; + var directServers = new List(); foreach (var identity in node.Identities) { if (TryResolveServerEndPoint(identity) is not { } server) continue; + if (Equals(server.EndPoint, identity) && !directServers.Contains(server)) directServers.Add(server); + } - if (survivor is null) - { - // Identities is ordered with the form the answering node advertised first, so the - // earliest match is also the deployment's stated preference - survivor = server; - } - else if (!ReferenceEquals(survivor, server)) + var survivor = ResolveKnownServer(node); + if (survivor is not null) + { + foreach (var server in directServers) { + if (ReferenceEquals(survivor, server)) continue; // ...unless one of them was configured: those may never be retired, so they win var pair = server.Provenance == ServerProvenance.Configured ? new DuplicateServers(server, survivor) @@ -1053,10 +1057,8 @@ internal async Task ApplyTopologyGenerationAsync(ClusterTopology topology, ILogg survivor = pair.Survivor; (merge ??= new()).Add(pair); } - } - if (survivor is not null) - { + SetPreferredDialEndPoint(node, survivor); survivor.OnSeenInTopology(generation); seen.Add(survivor); @@ -1141,6 +1143,7 @@ internal async Task RetireServerAsync(ServerEndPoint server, string reason, Time _serverIdentities.TryRemove(pair.Key, out _); } } + _preferredDialEndpoints.TryRemove(server, out _); } /// @@ -1152,23 +1155,59 @@ internal void RegisterServerIdentities(ClusterTopology topology) { foreach (var node in topology.Nodes) { - // resolve via any identity we already hold; if we know the node at all, the rest are aliases - ServerEndPoint? known = null; - foreach (var identity in node.Identities) + RegisterServerIdentities(node); + } + } + + private void RegisterServerIdentities(ClusterTopologyNode node) + { + if (ResolveKnownServer(node) is not { } known) return; + SetPreferredDialEndPoint(node, known); + + foreach (var identity in node.Identities) + { + if (servers[identity] is not null) continue; // already a server in its own right + var previous = _serverIdentities.AddOrUpdate(identity, known, (_, _) => known); + if (!ReferenceEquals(previous, known)) { - if ((known = TryResolveServerEndPoint(identity)) is not null) break; + Trace($"Identity {Format.ToString(identity)} -> {Format.ToString(known.EndPoint)}"); } - if (known is null) continue; + } + } - foreach (var identity in node.Identities) + private void SetPreferredDialEndPoint(ClusterTopologyNode node, ServerEndPoint server) + { + if (node.PreferredDialEndPoint is DnsEndPoint preferred) + { + _preferredDialEndpoints[server] = preferred; + } + else + { + _preferredDialEndpoints.TryRemove(server, out _); + } + } + + private ServerEndPoint? ResolveKnownServer(ClusterTopologyNode node) + { + var preferred = node.PreferredDialEndPoint; + ServerEndPoint? preferredKnown = null, dnsKnown = null, anyDirectKnown = null, aliasKnown = null; + + foreach (var identity in node.Identities) + { + if (TryResolveServerEndPoint(identity) is not { } known) continue; + if (!Equals(known.EndPoint, identity)) { - if (servers[identity] is not null) continue; // already a server in its own right - if (_serverIdentities.TryAdd(identity, known)) - { - Trace($"Identity {Format.ToString(identity)} -> {Format.ToString(known.EndPoint)}"); - } + aliasKnown ??= known; + continue; } + + anyDirectKnown ??= known; + if (preferred is not null && Equals(known.EndPoint, preferred)) preferredKnown = known; + if (known.EndPoint is DnsEndPoint) dnsKnown ??= known; } + + return preferredKnown + ?? (preferred is DnsEndPoint ? dnsKnown ?? anyDirectKnown ?? aliasKnown : anyDirectKnown ?? aliasKnown); } /// @@ -2249,6 +2288,8 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => var topology = ClusterTopology.From(slots); var clusterEndpoints = new EndPointCollection(); HashSet? listedInSlots = null; + Dictionary? preferredByIdentity = null; + Dictionary? nodeByIdentity = null; if (topology is not null) { @@ -2257,13 +2298,23 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => // another name is not duplicated RegisterServerIdentities(topology); listedInSlots = new HashSet(); + preferredByIdentity = new Dictionary(); + nodeByIdentity = new Dictionary(); foreach (var node in topology.Nodes) { if (SelectIdentity(node) is { } endpoint) clusterEndpoints.TryAdd(endpoint); // every name the node answers to, not just the one chosen above: NODES may well // report it under a different form than SLOTS did - foreach (var identity in node.Identities) listedInSlots.Add(identity); + foreach (var identity in node.Identities) + { + listedInSlots.Add(identity); + if (!nodeByIdentity.ContainsKey(identity)) nodeByIdentity.Add(identity, node); + if (node.PreferredDialEndPoint is { } preferred && !preferredByIdentity.ContainsKey(identity)) + { + preferredByIdentity.Add(identity, preferred); + } + } } } @@ -2286,12 +2337,17 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => else if (listedInSlots!.Contains(node.EndPoint)) { // SLOTS lists it, so it is in clusterEndpoints and we are about to connect to it and - // wait for it. Only the *existence* of a server was ever checked here, which a node - // in this state fails - a replica of a shard we have not reached yet is created by - // its primary's genealogy pass, and that has not run. Registering it inert then left - // it waited on but undialled, for the whole ConnectTimeout (#3232) - log?.LogInformationRegisteringSlotMapNode(new(node.EndPoint)); - GetServerEndPoint(node.EndPoint, ServerProvenance.ClusterTopology, log); + // wait for it. Prefer the SLOTS hostname over its companion IP from NODES: that IP + // may be an unreachable pod address using the ingress client port. + var dialEndPoint = preferredByIdentity!.TryGetValue(node.EndPoint, out var preferred) + ? preferred + : node.EndPoint; + log?.LogInformationRegisteringSlotMapNode(new(dialEndPoint)); + GetServerEndPoint(dialEndPoint, ServerProvenance.ClusterTopology, log); + if (!Equals(dialEndPoint, node.EndPoint) && nodeByIdentity!.TryGetValue(node.EndPoint, out var slotsNode)) + { + RegisterServerIdentities(slotsNode); + } } else { @@ -2320,15 +2376,12 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) => return null; } - // an identity we already hold if there is one - otherwise the form the answering node advertised, - // which is what Identities is ordered by, and which is the form TLS can validate EndPoint? SelectIdentity(ClusterTopologyNode node) { - foreach (var identity in node.Identities) - { - if (TryResolveServerEndPoint(identity) is { } known) return known.EndPoint; - } - return node.Identities.Count > 0 ? node.Identities[0] : null; + var known = ResolveKnownServer(node); + if (known is null) return node.PreferredDialEndPoint; + SetPreferredDialEndPoint(node, known); + return known.EndPoint; } } @@ -2552,8 +2605,6 @@ internal void UpdateClusterRange(ClusterTopology topology) { if (node.IsReplica || node.Slots.Count == 0) continue; - // resolve by *any* identity the node answers to before falling back to the form this reply - // happened to use; that is what keeps one node from becoming two ServerEndPoints var server = ResolveOrCreate(node); if (server is null) continue; @@ -2565,15 +2616,17 @@ internal void UpdateClusterRange(ClusterTopology topology) ServerEndPoint? ResolveOrCreate(ClusterTopologyNode node) { - foreach (var identity in node.Identities) + var server = ResolveKnownServer(node); + if (server is null && node.PreferredDialEndPoint is { } preferred) { - if (TryResolveServerEndPoint(identity) is { } known) return known; + server = GetServerEndPoint(preferred, ServerProvenance.ClusterTopology); } - - // unknown node: dial the form the answering node *advertised*, which Identities is ordered by. - // Not the address by preference: a certificate validates against a name, and where hostnames - // are preferred the advertised address may not even be routable (#2826) - return node.Identities.Count > 0 ? GetServerEndPoint(node.Identities[0], ServerProvenance.ClusterTopology) : null; + if (server is not null) + { + SetPreferredDialEndPoint(node, server); + RegisterServerIdentities(node); + } + return server; } } diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 9e5eab6a1..7337d35cb 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -296,7 +296,9 @@ private static void DiscardTimeout(ref CancellationTokenSource? source) internal async Task BeginConnectAsync(ILogger? log) { var bridge = BridgeCouldBeNull; - var endpoint = bridge?.ServerEndPoint?.EndPoint; + var endpoint = bridge is { } currentBridge && currentBridge.ServerEndPoint is { } serverEndPoint + ? currentBridge.Multiplexer.GetDialEndPoint(serverEndPoint) + : null; if (bridge == null || endpoint == null) { log?.LogErrorNoEndpoint(new ArgumentNullException(nameof(endpoint))); @@ -309,9 +311,8 @@ internal async Task BeginConnectAsync(ILogger? log) var connectTo = endpoint; // A MOVING may name where to go next; prefer it over resolving the endpoint again, since not - // waiting for DNS is the point. Note this changes only the *socket target*: the endpoint itself is - // untouched, so identity, server selection and - importantly - the TLS host and SNI (derived from - // ServerEndPoint.EndPoint below) all stay exactly as configured. + // waiting for DNS is the point. It changes only the socket target; the preferred dial endpoint + // remains the source for TLS host and SNI. if (bridge.ServerEndPoint?.HandoffTarget is { } handoffTarget) { Trace($"handoff: connecting to {Format.ToString(handoffTarget)} in place of {Format.ToString(endpoint)}"); @@ -1308,7 +1309,7 @@ internal async ValueTask ConnectedAsync(Socket? socket, ILogger? log) if (tunnel is not null) { _attemptStage = ConnectionAttemptStage.Tunnel; - stream = await tunnel.BeforeAuthenticateAsync(bridge.ServerEndPoint.EndPoint, bridge.ConnectionType, socket, CancellationToken.None).ForAwait(); + stream = await tunnel.BeforeAuthenticateAsync(bridge.Multiplexer.GetDialEndPoint(bridge.ServerEndPoint), bridge.ConnectionType, socket, CancellationToken.None).ForAwait(); } static Stream DemandSocketStream(Socket? socket) @@ -1318,7 +1319,7 @@ static Stream DemandSocketStream(Socket? socket) { _attemptStage = ConnectionAttemptStage.Tls; log?.LogInformationConfiguringTLS(); - var host = _tlsHostName = config.ResolveTlsHostName(bridge.ServerEndPoint.EndPoint); + var host = _tlsHostName = config.ResolveTlsHostName(bridge.Multiplexer.GetDialEndPoint(bridge.ServerEndPoint)); stream ??= DemandSocketStream(socket); var certificateValidation = config.CertificateValidationCallback ?? GetAmbientIssuerCertificateCallback(); diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index e203139c6..0262b8f79 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -532,9 +532,8 @@ internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge fro public RedisFeatures GetFeatures() => new RedisFeatures(version); /// - /// The CLUSTER SLOTS view of the topology, keyed on node-id. Populated alongside - /// but not yet used for routing, so the two can be compared - /// before anything depends on this one. + /// The CLUSTER SLOTS view of the topology, keyed on node-id. It drives slot routing when + /// available; continues to provide node relations. /// internal ClusterTopology? ClusterTopology { get; private set; } @@ -544,10 +543,10 @@ internal void SetClusterSlots(ClusterSlotsResult? slots) if (topology is not null) { ClusterTopology = topology; - Multiplexer.Trace($"Shadow topology: {topology.Nodes.Count} nodes"); + Multiplexer.Trace($"Cluster topology: {topology.Nodes.Count} nodes"); - // not routing on this yet, but the identities are useful immediately: they let a node reached - // by its other name resolve to the server we already have, rather than becoming a duplicate + // Register identities immediately, so a node reached by its other name resolves to the + // existing server rather than becoming a duplicate. Multiplexer.RegisterServerIdentities(topology); } } diff --git a/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs index dae5641e9..7ec7efd03 100644 --- a/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/ClusterTopologyUnitTests.cs @@ -101,6 +101,22 @@ public async Task ShadowTopologyKnowsBothIdentities(ClusterEndpointType preferre Assert.Contains(new DnsEndPoint(Hostname, port), node.Identities); } + [Theory] + [InlineData(ClusterEndpointType.Ip)] + [InlineData(ClusterEndpointType.Hostname)] + public async Task PreferredDialEndPointFollowsAnnouncedForm(ClusterEndpointType preferred) + { + using var server = CreateServer(log, preferred); + await using var conn = await server.ConnectAsync(defaultOnly: true); + + var node = Assert.Single((await GetShadowAsync(conn, server.DefaultEndPoint)).Nodes); + GetHost(server.DefaultEndPoint, out var port); + EndPoint expected = preferred == ClusterEndpointType.Hostname + ? new DnsEndPoint(Hostname, port) + : new IPEndPoint(IPAddress.Loopback, port); + Assert.Equal(expected, node.PreferredDialEndPoint); + } + [Fact] public async Task ShadowTopologyAgreesWithTheRoutingView() { @@ -172,20 +188,44 @@ public async Task SlotMapIsDrivenByTheSlotsView() [Fact] public async Task HostnamePreferredClusterRoutesWithoutDuplicatingEndpoints() + => await RunHostnamePreferredClusterRoutesWithoutDuplicatingEndpoints(defaultOnly: false); + + [Fact] + public async Task HostnamePreferredClusterDiscoversFromSingleSeed() + => await RunHostnamePreferredClusterRoutesWithoutDuplicatingEndpoints(defaultOnly: true); + + private async Task RunHostnamePreferredClusterRoutesWithoutDuplicatingEndpoints(bool defaultOnly) { - // the case the flip exists for: SLOTS names every node by hostname while NODES names them by - // address, so a slot map fed from SLOTS must still resolve to the servers we already hold + // SLOTS names the shard by hostname, while NODES reports its companion IP. Cover both already-seeded + // endpoints and discovery from a single seed: either path must keep one server and dial with the hostname. using var server = CreateServer(log, ClusterEndpointType.Hostname); GetHost(server.DefaultEndPoint, out var port); var other = server.AddEmptyNode(new IPEndPoint(IPAddress.Loopback, port + 1)); server.SetHostname(other, "host-2.redis.example.com"); server.Migrate((RedisKey)"hostname-routed-key", other); - await using var conn = await server.ConnectAsync(); + await using var conn = await server.ConnectAsync(defaultOnly: defaultOnly); var db = conn.GetDatabase(); await db.StringSetAsync("hostname-routed-key", "value"); Assert.Equal("value", await db.StringGetAsync("hostname-routed-key")); + var hostnameEndpoint = new DnsEndPoint("host-2.redis.example.com", port + 1); + var companionIp = new IPEndPoint(IPAddress.Loopback, port + 1); + if (defaultOnly) + { + Assert.Contains(hostnameEndpoint, conn.GetEndPoints()); + Assert.DoesNotContain(companionIp, conn.GetEndPoints()); + } + else + { + Assert.Contains(companionIp, conn.GetEndPoints()); + Assert.DoesNotContain(hostnameEndpoint, conn.GetEndPoints()); + } + var multiplexer = Assert.IsType(conn); + var serverEndPoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(companionIp); + Assert.Equal(hostnameEndpoint, multiplexer.GetDialEndPoint(serverEndPoint)); + Assert.Equal(serverEndPoint, ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(hostnameEndpoint)); + foreach (var ep in conn.GetEndPoints()) { log.WriteLine($"endpoint: {ep}");