Ticket #102 : When node is stopped then it become inactive

habarthierry-hue · habarthierry-hue · commit 31d220d5f875 · 2022-05-10T16:02:43.000+02:00
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/Extensions/AsyncExtensions.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/Extensions/AsyncExtensions.cs
@@ -16,7 +16,6 @@ public static async Task<T> WithCancellation<T>(this Task<T> task, CancellationT
                 {
                     task, tcs.Task
                 };
-
                 Task delayTask = null;
                 if (timeOutMilliSeconds != null)
                 {
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/GossipClient.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/GossipClient.cs
@@ -25,30 +25,31 @@ public GossipClient(string url, int port) : this(new IPEndPoint(IPAddressHelper.
 
         public UdpClient UdpClient { get; private set; }
 
-        public Task Heartbeat(string url, int port, CancellationToken cancellationToken = default(CancellationToken))
+        public Task Heartbeat(string url, int port, int? timeout, CancellationToken cancellationToken = default(CancellationToken))
         {
             var package = GossipPackageRequestBuilder.Heartbeat(url, port);
-            return Send(package, cancellationToken);
-        }
-
-        public async Task Send(GossipPackage gossipPackage, CancellationToken cancellationToken = default(CancellationToken))
-        {
-            var writeCtx = new WriteBufferContext();
-            gossipPackage.Serialize(writeCtx);
-            var payload = writeCtx.Buffer.ToArray();
-            await UdpClient.SendAsync(payload, payload.Count(), _target).WithCancellation(cancellationToken);
+            return Send(package, timeout, cancellationToken);
         }
 
         public Task JoinNode(string url, int port, CancellationToken cancellationToken = default(CancellationToken))
         {
             var package = GossipPackageRequestBuilder.AddNode(url, port);
-            return Send(package, cancellationToken);
+            return Send(package, cancellationToken: cancellationToken);
         }
 
         public Task UpdateClusterNodes(string url, int port, ICollection<ClusterNodeMessage> clusterNodes, CancellationToken cancellationToken = default(CancellationToken))
         {
             var package = GossipPackageRequestBuilder.UpdateClusterNodes(url, port, clusterNodes);
-            return Send(package, cancellationToken);
+            return Send(package, cancellationToken: cancellationToken);
+        }
+
+        public async Task Send(GossipPackage gossipPackage, int? timeoutMS = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var writeCtx = new WriteBufferContext();
+            gossipPackage.Serialize(writeCtx);
+            var payload = writeCtx.Buffer.ToArray();
+            await UdpClient.SendAsync(payload, payload.Count(), _target).WithCancellation(cancellationToken, timeoutMS);
+            if (UdpClient.Available == 1) throw new TimeoutException();
         }
 
         public void Dispose()
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs
@@ -25,6 +25,7 @@ public interface INodeHost
         int Port { get; }
         bool IsStarted { get; }
         BlockingCollection<IPeerHost> Peers { get; }
+        BlockingCollection<UnreachableClusterNode> UnreachableClusterNodes { get; }
         INodeStateStore NodeStateStore { get; }
         Task Start(CancellationToken cancellationToken);
         Task Stop();
@@ -42,6 +43,7 @@ public abstract class BaseNodeHost : INodeHost
         private BlockingCollection<IPeerHost> _peers;
         private string _nodeId;
         private bool _isStarted = false;
+        private BlockingCollection<UnreachableClusterNode> _clusterNodes;
 
         public BaseNodeHost(IPeerStore peerStore, IPeerHostFactory peerHostFactory, INodeStateStore nodeStateStore, IClusterStore clusterStore, ILogger<BaseNodeHost> logger, IOptions<ConsensusPeerOptions> options)
         {
@@ -56,7 +58,9 @@ public BaseNodeHost(IPeerStore peerStore, IPeerHostFactory peerHostFactory, INod
         }
 
         public BlockingCollection<IPeerHost> Peers => _peers;
+        public BlockingCollection<UnreachableClusterNode> UnreachableClusterNodes => _clusterNodes;
         public INodeStateStore NodeStateStore => _nodeStateStore;
+        public IClusterStore ClusterStore => _clusterStore;
         public bool IsStarted => _isStarted;
         public int Port => _options.Port;
         public bool IsRunning { get; private set; }
@@ -71,6 +75,7 @@ public async Task Start(CancellationToken cancellationToken)
         {
             if (IsRunning) throw new InvalidOperationException("The node is already running");
             _peers = new BlockingCollection<IPeerHost>();
+            _clusterNodes = new BlockingCollection<UnreachableClusterNode>();
             TokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
             await RunConsensusPeers(cancellationToken);
             StartGossipTimer();
@@ -174,7 +179,7 @@ private async Task<bool> TryHandleGossipRequest(UdpReceiveResult udpResult)
             if(packageResult == null) return true;
             using(var client = new GossipClient(gossipPackage.Header.SourceUrl, gossipPackage.Header.SourcePort))
             {
-                await client.Send(packageResult, TokenSource.Token);
+                await client.Send(packageResult, cancellationToken: TokenSource.Token);
             }
 
             return true;
@@ -205,9 +210,20 @@ private async Task BroadcastGossipHeartbeat(object source, ElapsedEventArgs e)
             {
                 var rndNodeIndex = rnd.Next(0, totalNodes);
                 var selectedNode = nodes.ElementAt(rndNodeIndex);
-                using (var client = new GossipClient(selectedNode.Url, selectedNode.Port))
+                if (_clusterNodes.Any(n => n.Node.Url == selectedNode.Url && n.Node.Port == selectedNode.Port && n.ReactivationDateTime > DateTime.UtcNow)) continue;
+                try
                 {
-                    await client.Heartbeat(_options.Url, _options.Port, TokenSource.Token);
+                    using (var client = new GossipClient(selectedNode.Url, selectedNode.Port))
+                    {
+                        await client.Heartbeat(_options.Url, _options.Port, _options.GossipTimeoutHeartbeatMS, TokenSource.Token);
+                    }
+                }
+                catch(Exception ex)
+                {
+                    Logger.LogError(ex.ToString());
+                    var clusterNode = _clusterNodes.FirstOrDefault(c => c.Node.Port == selectedNode.Port && c.Node.Url == selectedNode.Url);
+                    if (clusterNode == null) _clusterNodes.Add(new UnreachableClusterNode(selectedNode, DateTime.UtcNow.AddMilliseconds(_options.GossipClusterNodeDeactivationDurationMS)));
+                    else clusterNode.ReactivationDateTime = DateTime.UtcNow.AddMilliseconds(_options.GossipClusterNodeDeactivationDurationMS);
                 }
             }
 
@@ -289,4 +305,16 @@ private async Task<GossipPackage> HandleGossipRequest(GossipUpdateClusterRequest
 
         #endregion
     }
+
+    public class UnreachableClusterNode
+    {
+        public UnreachableClusterNode(ClusterNode node, DateTime reactivationDateTime)
+        {
+            Node = node;
+            ReactivationDateTime = reactivationDateTime;
+        }
+
+        public ClusterNode Node { get; private set; }
+        public DateTime ReactivationDateTime { get; set; }
+    }
 }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/ConsensusPeerOptions.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/ConsensusPeerOptions.cs
@@ -16,6 +16,8 @@ public ConsensusPeerOptions()
             LeaderHeartbeatTimerMS = 1000;
             GossipTimerMS = 1000;
             GossipMaxNodeBroadcast = 2;
+            GossipTimeoutHeartbeatMS = 2000;
+            GossipClusterNodeDeactivationDurationMS = 3000;
         }
 
         /// <summary>
@@ -55,9 +57,17 @@ public ConsensusPeerOptions()
         /// </summary>
         public int GossipTimerMS { get; set; }
         /// <summary>
-        /// Maximum number of nodes to broadcast the message.
+        /// Gossip - Maximum number of nodes to broadcast the message.
         /// </summary>
         public int GossipMaxNodeBroadcast{ get; set; }
+        /// <summary>
+        /// Gossip - Heartbeat request expire after MS.
+        /// </summary>
+        public int GossipTimeoutHeartbeatMS { get; set; }
+        /// <summary>
+        /// Gossip - When cluster node is not reachable then deactivate the node.
+        /// </summary>
+        public int GossipClusterNodeDeactivationDurationMS { get; set; }
     }
 
     public class Interval
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/ClusterStore.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/ClusterStore.cs
@@ -1,6 +1,5 @@
 ﻿using FaasNet.RaftConsensus.Client;
 using FaasNet.RaftConsensus.Core.Models;
-using Microsoft.Extensions.Options;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text.Json;
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs
@@ -104,7 +104,7 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
     var continueExecution = true;
     do
     {
-        Console.WriteLine("Enter 'Q' to stop execution / Enter 'inf' to display the states / Start a new node 'addnode'");
+        Console.WriteLine("Enter 'Q' to stop execution / Enter 'inf' to display the states / Start a new node 'addnode' / Stop one node 'stopnode'");
         string termId = Console.ReadLine();
         continueExecution = termId != "Q";
         if (termId == "inf")
@@ -130,6 +130,15 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
                 false);
             await StartNode(newNode, nodes.Count());
             nodes.Add(newNode);
+            continue;
+        }
+
+        if (termId == "stopnode")
+        {
+            Console.WriteLine("Enter the index");
+            int index = int.Parse(Console.ReadLine());
+            await nodes.ElementAt(index).Stop();
+            continue;
         }
     }
     while (continueExecution);
diff --git a/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs b/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs
@@ -31,6 +31,8 @@ public async Task When_JoinNodeToCluster_Then_StorageIsUpdated()
             using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.JoinNode("localhost", 4001);
             var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
             var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
+            await seedNode.Stop();
+            await firstNode.Stop();
 
             // ASSERT
             Assert.Equal(2, seedNodeStates.Count());
@@ -43,6 +45,31 @@ public async Task When_JoinNodeToCluster_Then_StorageIsUpdated()
             Assert.Equal("{\"Url\":\"localhost\",\"Port\":4001}", firstNodeStates.ElementAt(0).Value);
         }
 
+        [Fact]
+        public async Task When_NodeIsStopped_Then_NodeBecomeUnreachable()
+        {
+            // ARRANGE
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4000, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4001, new ConcurrentBag<ClusterNode>());
+            await seedNode.Start(CancellationToken.None);
+            await firstNode.Start(CancellationToken.None);
+            WaitNodeIsStarted(seedNode);
+            WaitNodeIsStarted(firstNode);
+            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.JoinNode("localhost", 4001);
+            var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
+            var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
+
+            // ACT
+            await firstNode.Stop();
+            var clusterNodes = WaitUnreachableClusterNodes(seedNode, (un) => un.Count() == 1);
+            await seedNode.Stop();
+
+            // ASSERT
+            Assert.Single(clusterNodes);
+            Assert.Equal("localhost", clusterNodes.First().Node.Url);
+            Assert.Equal(4001, clusterNodes.First().Node.Port);
+        }
+
         #endregion
 
         [Fact]
@@ -175,6 +202,15 @@ private static async Task<IEnumerable<NodeState>> WaitEntityTypes(INodeHost node
             }
         }
 
+        private static IEnumerable<UnreachableClusterNode> WaitUnreachableClusterNodes(INodeHost node, Func<IEnumerable<UnreachableClusterNode>, bool> callback)
+        {
+            while(true)
+            {
+                if (callback(node.UnreachableClusterNodes)) return node.UnreachableClusterNodes.ToArray();
+                Thread.Sleep(200);
+            }
+        }
+
         private static void WaitOnlyOneLeader(List<INodeHost> nodes, string termId)
         {
             while (true)

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@ public static async Task<T> WithCancellation<T>(this Task<T> task, CancellationT`
`16`	`16`	`{`
`17`	`17`	`task, tcs.Task`
`18`	`18`	`};`
`19`		`-`
`20`	`19`	`Task delayTask = null;`
`21`	`20`	`if (timeOutMilliSeconds != null)`
`22`	`21`	`{`