[Membership] Ignore stale silos when picking a probing intermediary (#9335)

Author: ReubenBond

src/Orleans.Runtime/MembershipService/ClusterHealthMonitor.cs

@@ -323,12 +323,12 @@ private async Task OnProbeResultInternal(SiloHealthMonitor monitor, ProbeResult
             {
                 if (probeResult.Status == ProbeResultStatus.Failed && probeResult.FailedProbeCount >= this.clusterMembershipOptions.CurrentValue.NumMissedProbesLimit)
                 {
-                    await this.membershipService.TryToSuspectOrKill(monitor.SiloAddress).ConfigureAwait(false);
+                    await this.membershipService.TryToSuspectOrKill(monitor.TargetSiloAddress).ConfigureAwait(false);
                 }
             }
             else if (probeResult.Status == ProbeResultStatus.Failed)
             {
-                await this.membershipService.TryToSuspectOrKill(monitor.SiloAddress, probeResult.Intermediary).ConfigureAwait(false);
+                await this.membershipService.TryToSuspectOrKill(monitor.TargetSiloAddress, probeResult.Intermediary).ConfigureAwait(false);
             }
         }
 
@@ -341,7 +341,7 @@ bool IHealthCheckable.CheckHealth(DateTime lastCheckTime, out string reason)
                 ok &= monitor.CheckHealth(lastCheckTime, out var monitorReason);
                 if (!string.IsNullOrWhiteSpace(monitorReason))
                 {
-                    var siloReason = $"Monitor for {monitor.SiloAddress} is degraded with: {monitorReason}.";
+                    var siloReason = $"Monitor for {monitor.TargetSiloAddress} is degraded with: {monitorReason}.";
                     if (string.IsNullOrWhiteSpace(reason))
                     {
                         reason = siloReason;
@@ -375,7 +375,7 @@ public void Dispose()
                 }
                 catch (Exception exception)
                 {
-                    log.LogError(exception, "Error disposing monitor for {SiloAddress}.", monitor.SiloAddress);
+                    log.LogError(exception, "Error disposing monitor for {SiloAddress}.", monitor.TargetSiloAddress);
                 }
             }
         }
@@ -400,7 +400,7 @@ public async ValueTask DisposeAsync()
                 }
                 catch (Exception exception)
                 {
-                    log.LogError(exception, "Error disposing monitor for {SiloAddress}.", monitor.SiloAddress);
+                    log.LogError(exception, "Error disposing monitor for {SiloAddress}.", monitor.TargetSiloAddress);
                 }
             }
 

src/Orleans.Runtime/MembershipService/SiloHealthMonitor.cs

@@ -22,7 +22,7 @@ internal class SiloHealthMonitor : ITestAccessor, IHealthCheckable, IDisposable,
         private readonly IOptionsMonitor<ClusterMembershipOptions> _clusterMembershipOptions;
         private readonly IRemoteSiloProber _prober;
         private readonly ILocalSiloHealthMonitor _localSiloHealthMonitor;
-        private readonly IClusterMembershipService _membershipService;
+        private readonly MembershipTableManager _membershipService;
         private readonly ILocalSiloDetails _localSiloDetails;
         private readonly CancellationTokenSource _stoppingCancellation = new();
         private readonly object _lockObj = new();
@@ -59,10 +59,10 @@ public SiloHealthMonitor(
             IRemoteSiloProber remoteSiloProber,
             IAsyncTimerFactory asyncTimerFactory,
             ILocalSiloHealthMonitor localSiloHealthMonitor,
-            IClusterMembershipService membershipService,
+            MembershipTableManager membershipService,
             ILocalSiloDetails localSiloDetails)
         {
-            SiloAddress = siloAddress;
+            TargetSiloAddress = siloAddress;
             _clusterMembershipOptions = clusterMembershipOptions;
             _prober = remoteSiloProber;
             _localSiloHealthMonitor = localSiloHealthMonitor;
@@ -84,7 +84,7 @@ internal interface ITestAccessor
         /// <summary>
         /// The silo which this instance is responsible for.
         /// </summary>
-        public SiloAddress SiloAddress { get; }
+        public SiloAddress TargetSiloAddress { get; }
 
         /// <summary>
         /// Whether or not this monitor is canceled.
@@ -153,28 +153,33 @@ public async ValueTask DisposeAsync()
 
         private async Task Run()
         {
-            ClusterMembershipSnapshot? activeMembersSnapshot = default;
+            MembershipTableSnapshot? activeMembersSnapshot = default;
             SiloAddress[]? otherNodes = default;
-            TimeSpan? overrideDelay = RandomTimeSpan.Next(_clusterMembershipOptions.CurrentValue.ProbeTimeout);
+            var options = _clusterMembershipOptions.CurrentValue;
+            TimeSpan? overrideDelay = RandomTimeSpan.Next(options.ProbeTimeout);
             while (await _pingTimer.NextTick(overrideDelay))
             {
                 ProbeResult probeResult;
                 overrideDelay = default;
+                var now = DateTime.UtcNow;
 
                 try
                 {
                     // Discover the other active nodes in the cluster, if there are any.
-                    var membershipSnapshot = _membershipService.CurrentSnapshot;
+                    var membershipSnapshot = _membershipService.MembershipTableSnapshot;
                     if (otherNodes is null || !ReferenceEquals(activeMembersSnapshot, membershipSnapshot))
                     {
                         activeMembersSnapshot = membershipSnapshot;
-                        otherNodes = membershipSnapshot.Members.Values
-                            .Where(v => v.Status == SiloStatus.Active && v.SiloAddress != this.SiloAddress && v.SiloAddress != _localSiloDetails.SiloAddress)
+                        otherNodes = membershipSnapshot.Entries.Values
+                            .Where(v => v.Status == SiloStatus.Active
+                                && !v.SiloAddress.Equals(TargetSiloAddress)
+                                && !v.SiloAddress.Equals(_localSiloDetails.SiloAddress)
+                                && !v.HasMissedIAmAlives(options, now))
                             .Select(s => s.SiloAddress)
                             .ToArray();
                     }
 
-                    var isDirectProbe = !_clusterMembershipOptions.CurrentValue.EnableIndirectProbes || _failedProbes < _clusterMembershipOptions.CurrentValue.NumMissedProbesLimit - 1 || otherNodes.Length == 0;
+                    var isDirectProbe = !options.EnableIndirectProbes || _failedProbes < options.NumMissedProbesLimit - 1 || otherNodes.Length == 0;
                     var timeout = GetTimeout(isDirectProbe);
                     using var cancellation = new CancellationTokenSource(timeout);
 
@@ -198,7 +203,7 @@ private async Task Run()
                         if (probeResult.Status != ProbeResultStatus.Succeeded && probeResult.IntermediaryHealthDegradationScore > 0)
                         {
                             _log.LogInformation("Recusing unhealthy intermediary '{Intermediary}' and trying again with remaining nodes", intermediary);
-                            otherNodes = otherNodes.Where(node => !node.Equals(intermediary)).ToArray();
+                            otherNodes = [.. otherNodes.Where(node => !node.Equals(intermediary))];
                             overrideDelay = TimeSpan.FromMilliseconds(250);
                         }
                     }
@@ -210,15 +215,15 @@ private async Task Run()
                 }
                 catch (Exception exception)
                 {
-                    _log.LogError(exception, "Exception monitoring silo {SiloAddress}", SiloAddress);
+                    _log.LogError(exception, "Exception monitoring silo {SiloAddress}", TargetSiloAddress);
                 }
             }
 
             TimeSpan GetTimeout(bool isDirectProbe)
             {
                 var additionalTimeout = 0;
 
-                if (_clusterMembershipOptions.CurrentValue.ExtendProbeTimeoutDuringDegradation)
+                if (options.ExtendProbeTimeoutDuringDegradation)
                 {
                     // Attempt to account for local health degradation by extending the timeout period.
                     var localDegradationScore = _localSiloHealthMonitor.GetLocalHealthDegradationScore(DateTime.UtcNow);
@@ -231,7 +236,7 @@ TimeSpan GetTimeout(bool isDirectProbe)
                     additionalTimeout += 1;
                 }
 
-                return _clusterMembershipOptions.CurrentValue.ProbeTimeout.Multiply(1 + additionalTimeout);
+                return options.ProbeTimeout.Multiply(1 + additionalTimeout);
             }
         }
 
@@ -245,15 +250,15 @@ private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
             var id = ++_nextProbeId;
             if (_log.IsEnabled(LogLevel.Trace))
             {
-                _log.LogTrace("Going to send Ping #{Id} to probe silo {Silo}", id, SiloAddress);
+                _log.LogTrace("Going to send Ping #{Id} to probe silo {Silo}", id, TargetSiloAddress);
             }
 
             var roundTripTimer = ValueStopwatch.StartNew();
             ProbeResult probeResult;
             Exception? failureException;
             try
             {
-                await _prober.Probe(SiloAddress, id, cancellation).WaitAsync(cancellation);
+                await _prober.Probe(TargetSiloAddress, id, cancellation).WaitAsync(cancellation);
                 failureException = null;
             }
             catch (OperationCanceledException exception)
@@ -271,14 +276,14 @@ private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
 
             if (failureException is null)
             {
-                MessagingInstruments.OnPingReplyReceived(SiloAddress);
+                MessagingInstruments.OnPingReplyReceived(TargetSiloAddress);
 
                 if (_log.IsEnabled(LogLevel.Trace))
                 {
                     _log.LogTrace(
                         "Got successful ping response for ping #{Id} from {Silo} with round trip time of {RoundTripTime}",
                         id,
-                        SiloAddress,
+                        TargetSiloAddress,
                         roundTripTimer.Elapsed);
                 }
 
@@ -289,15 +294,15 @@ private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
             }
             else
             {
-                MessagingInstruments.OnPingReplyMissed(SiloAddress);
+                MessagingInstruments.OnPingReplyMissed(TargetSiloAddress);
 
                 var failedProbes = ++_failedProbes;
                 _log.LogWarning(
                     (int)ErrorCode.MembershipMissedPing,
                     failureException,
                     "Did not get response for probe #{Id} to silo {Silo} after {Elapsed}. Current number of consecutive failed probes is {FailedProbeCount}",
                     id,
-                    SiloAddress,
+                    TargetSiloAddress,
                     roundTripTimer.Elapsed,
                     failedProbes);
 
@@ -319,15 +324,15 @@ private async Task<ProbeResult> ProbeIndirectly(SiloAddress intermediary, TimeSp
             var id = ++_nextProbeId;
             if (_log.IsEnabled(LogLevel.Trace))
             {
-                _log.LogTrace("Going to send indirect ping #{Id} to probe silo {Silo} via {Intermediary}", id, SiloAddress, intermediary);
+                _log.LogTrace("Going to send indirect ping #{Id} to probe silo {Silo} via {Intermediary}", id, TargetSiloAddress, intermediary);
             }
 
             var roundTripTimer = ValueStopwatch.StartNew();
             ProbeResult probeResult;
             try
             {
                 using var cancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellation, _stoppingCancellation.Token);
-                var indirectResult = await _prober.ProbeIndirectly(intermediary, SiloAddress, directProbeTimeout, id, cancellationSource.Token).WaitAsync(cancellationSource.Token);
+                var indirectResult = await _prober.ProbeIndirectly(intermediary, TargetSiloAddress, directProbeTimeout, id, cancellationSource.Token).WaitAsync(cancellationSource.Token);
                 roundTripTimer.Stop();
                 var roundTripTime = roundTripTimer.Elapsed - indirectResult.ProbeResponseTime;
 
@@ -340,26 +345,26 @@ private async Task<ProbeResult> ProbeIndirectly(SiloAddress intermediary, TimeSp
                     _log.LogInformation(
                         "Indirect probe request #{Id} to silo {SiloAddress} via silo {IntermediarySiloAddress} succeeded after {RoundTripTime} with a direct probe response time of {ProbeResponseTime}.",
                         id,
-                        SiloAddress,
+                        TargetSiloAddress,
                         intermediary,
                         roundTripTimer.Elapsed,
                         indirectResult.ProbeResponseTime);
 
-                    MessagingInstruments.OnPingReplyReceived(SiloAddress);
+                    MessagingInstruments.OnPingReplyReceived(TargetSiloAddress);
 
                     _failedProbes = 0;
                     probeResult = ProbeResult.CreateIndirect(0, ProbeResultStatus.Succeeded, indirectResult, intermediary);
                 }
                 else
                 {
-                    MessagingInstruments.OnPingReplyMissed(SiloAddress);
+                    MessagingInstruments.OnPingReplyMissed(TargetSiloAddress);
 
                     if (indirectResult.IntermediaryHealthScore > 0)
                     {
                         _log.LogInformation(
                             "Ignoring failure result for ping #{Id} from {Silo} since the intermediary used to probe the silo is not healthy. Intermediary health degradation score: {IntermediaryHealthScore}.",
                             id,
-                            SiloAddress,
+                            TargetSiloAddress,
                             indirectResult.IntermediaryHealthScore);
                         probeResult = ProbeResult.CreateIndirect(_failedProbes, ProbeResultStatus.Unknown, indirectResult, intermediary);
                     }
@@ -368,7 +373,7 @@ private async Task<ProbeResult> ProbeIndirectly(SiloAddress intermediary, TimeSp
                         _log.LogWarning(
                             "Indirect probe request #{Id} to silo {SiloAddress} via silo {IntermediarySiloAddress} failed after {RoundTripTime} with a direct probe response time of {ProbeResponseTime}. Failure message: {FailureMessage}. Intermediary health score: {IntermediaryHealthScore}.",
                             id,
-                            SiloAddress,
+                            TargetSiloAddress,
                             intermediary,
                             roundTripTimer.Elapsed,
                             indirectResult.ProbeResponseTime,
@@ -382,7 +387,7 @@ private async Task<ProbeResult> ProbeIndirectly(SiloAddress intermediary, TimeSp
             }
             catch (Exception exception)
             {
-                MessagingInstruments.OnPingReplyMissed(SiloAddress);
+                MessagingInstruments.OnPingReplyMissed(TargetSiloAddress);
                 _log.LogWarning(exception, "Indirect probe request failed.");
                 probeResult = ProbeResult.CreateIndirect(_failedProbes, ProbeResultStatus.Unknown, default, intermediary);
             }

test/NonSilo.Tests/Membership/ClusterHealthMonitorTests.cs

@@ -224,7 +224,7 @@ private async Task ClusterHealthMonitor_BasicScenario_Runner(bool enableIndirect
             Assert.DoesNotContain(testRig.TestAccessor.MonitoredSilos, s => s.Key.Equals(this.localSilo));
             var expectedNumProbedSilos = otherSilosAreStale ? otherSilos.Length : clusterMembershipOptions.NumProbedSilos;
             Assert.Equal(expectedNumProbedSilos, testRig.TestAccessor.MonitoredSilos.Count);
-            Assert.All(testRig.TestAccessor.MonitoredSilos, m => m.Key.Equals(m.Value.SiloAddress));
+            Assert.All(testRig.TestAccessor.MonitoredSilos, m => m.Key.Equals(m.Value.TargetSiloAddress));
             Assert.Empty(probeCalls);
 
             // Check that those silos are actually being probed periodically
@@ -290,7 +290,7 @@ await UntilEqual(expectedNumProbedSilos, () =>
                 {
                     Assert.Equal(expectedMissedProbes, ((SiloHealthMonitor.ITestAccessor)siloMonitor).MissedProbes);
 
-                    var entry = table.Members.Single(m => m.Item1.SiloAddress.Equals(siloMonitor.SiloAddress)).Item1;
+                    var entry = table.Members.Single(m => m.Item1.SiloAddress.Equals(siloMonitor.TargetSiloAddress)).Item1;
                     var votes = entry.GetFreshVotes(now.UtcDateTime, clusterMembershipOptions.DeathVoteExpirationTimeout);
                     if (expectedMissedProbes < clusterMembershipOptions.NumMissedProbesLimit)
                     {
@@ -359,7 +359,7 @@ await UntilEqual(testRig.TestAccessor.MonitoredSilos.Count, () =>
 
             foreach (var siloMonitor in testRig.TestAccessor.MonitoredSilos.Values)
             {
-                this.output.WriteLine($"Checking missed probes on {siloMonitor.SiloAddress}: {((SiloHealthMonitor.ITestAccessor)siloMonitor).MissedProbes}");
+                this.output.WriteLine($"Checking missed probes on {siloMonitor.TargetSiloAddress}: {((SiloHealthMonitor.ITestAccessor)siloMonitor).MissedProbes}");
                 Assert.Equal(0, ((SiloHealthMonitor.ITestAccessor)siloMonitor).MissedProbes);
             }
 
@@ -533,12 +533,10 @@ private async Task StopLifecycle(CancellationToken cancellation = default)
 
         private class ClusterHealthMonitorTestRig(
             MembershipTableManager manager,
-            IClusterMembershipService membershipService,
             IOptionsMonitor<ClusterMembershipOptions> optionsMonitor,
             ClusterHealthMonitor.ITestAccessor testAccessor)
         {
             public readonly MembershipTableManager Manager = manager;
-            public readonly IClusterMembershipService MembershipService = membershipService;
             public readonly IOptionsMonitor<ClusterMembershipOptions> OptionsMonitor = optionsMonitor;
             public readonly ClusterHealthMonitor.ITestAccessor TestAccessor = testAccessor;
         }
@@ -557,9 +555,6 @@ private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMem
 
             ((ILifecycleParticipant<ISiloLifecycle>)manager).Participate(this.lifecycle);
 
-            var membershipService = Substitute.For<IClusterMembershipService>();
-            membershipService.CurrentSnapshot.ReturnsForAnyArgs(info => manager.MembershipTableSnapshot.CreateClusterMembershipSnapshot());
-
             var optionsMonitor = Substitute.For<IOptionsMonitor<ClusterMembershipOptions>>();
             optionsMonitor.CurrentValue.ReturnsForAnyArgs(clusterMembershipOptions);
 
@@ -582,12 +577,11 @@ private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMem
                 this.prober,
                 this.timerFactory,
                 this.localSiloHealthMonitor,
-                membershipService,
+                manager,
                 this.localSiloDetails);
 
             return new(
                 manager: manager,
-                membershipService: membershipService,
                 optionsMonitor: optionsMonitor,
                 testAccessor: testAccessor);
         }