Ticket #102 : Propagate cluster nodes.

habarthierry-hue · habarthierry-hue · commit 210421e469ac · 2022-05-10T14:55:08.000+02:00
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/GossipClient.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/GossipClient.cs
@@ -2,6 +2,7 @@
 using FaasNet.RaftConsensus.Client.Messages;
 using FaasNet.RaftConsensus.Client.Messages.Gossip;
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
@@ -44,6 +45,12 @@ public GossipClient(string url, int port) : this(new IPEndPoint(IPAddressHelper.
             return Send(package, 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);
+        }
+
         public void Dispose()
         {
             UdpClient?.Dispose();
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipCommands.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipCommands.cs
@@ -8,6 +8,7 @@ public class GossipCommands: BaseCommands
         public static GossipCommands SYNC_RESULT = new GossipCommands(3, "SYNC_RESULT");
         public static GossipCommands UPDATE_NODE_STATE_REQUEST = new GossipCommands(4, "UPDATE_NODE_STATE_REQUEST");
         public static GossipCommands JOIN_NODE_REQUEST = new GossipCommands(5, "JOIN_NODE_REQUEST");
+        public static GossipCommands UPDATE_CLUSTER_NODES_REQUEST = new GossipCommands(6, "UPDATE_CLUSTER_NODES_REQUEST");
 
         protected GossipCommands(int code)
         {
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipPackage.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipPackage.cs
@@ -56,6 +56,13 @@ public static GossipPackage Deserialize(ReadBufferContext context)
                 return result;
             }
 
+            if (header.Command == GossipCommands.UPDATE_CLUSTER_NODES_REQUEST)
+            {
+                var result = new GossipUpdateClusterRequest { Header = header };
+                result.Extract(context);
+                return result;
+            }
+
             return new GossipPackage { Header = header };
         }
     }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipPackageRequestBuilder.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipPackageRequestBuilder.cs
@@ -31,5 +31,14 @@ public static GossipPackage AddNode(string url, int port)
                 Url = url
             };
         }
+
+        public static GossipPackage UpdateClusterNodes(string url, int port, ICollection<ClusterNodeMessage> clusterNodes)
+        {
+            return new GossipUpdateClusterRequest
+            {
+                Header = new GossipHeader(GossipCommands.UPDATE_CLUSTER_NODES_REQUEST, url, port),
+                Nodes = clusterNodes
+            };
+        }
     }
 }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipUpdateClusterRequest.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipUpdateClusterRequest.cs
@@ -1,6 +1,48 @@
-﻿namespace FaasNet.RaftConsensus.Client.Messages.Gossip
+﻿using System.Collections.Generic;
+
+namespace FaasNet.RaftConsensus.Client.Messages.Gossip
 {
-    public class GossipUpdateClusterRequest
+    public class GossipUpdateClusterRequest : GossipPackage
     {
+        public GossipUpdateClusterRequest()
+        {
+            Nodes = new List<ClusterNodeMessage>();
+        }
+
+        public ICollection<ClusterNodeMessage> Nodes { get; set; }
+
+        public override void Serialize(WriteBufferContext context)
+        {
+            base.Serialize(context);
+            context.WriteInteger(Nodes.Count);
+            foreach (var node in Nodes) node.Serialize(context);
+        }
+
+        public void Extract(ReadBufferContext context)
+        {
+            var size = context.NextInt();
+            for(int i = 0; i < size; i++) Nodes.Add(ClusterNodeMessage.Deserialize(context));
+        }
+    }
+
+    public class ClusterNodeMessage
+    {
+        public string Url { get; set; }
+        public int Port { get; set; }
+
+        public void Serialize(WriteBufferContext context)
+        {
+            context.WriteString(Url);
+            context.WriteInteger(Port);
+        }
+
+        public static ClusterNodeMessage Deserialize(ReadBufferContext context)
+        {
+            return new ClusterNodeMessage
+            {
+                Url = context.NextString(),
+                Port = context.NextInt()
+            };
+        }
     }
 }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs
@@ -22,7 +22,10 @@ public interface INodeHost
     {
         event EventHandler<EventArgs> NodeStarted;
         string NodeId { get; }
+        int Port { get; }
+        bool IsStarted { get; }
         BlockingCollection<IPeerHost> Peers { get; }
+        INodeStateStore NodeStateStore { get; }
         Task Start(CancellationToken cancellationToken);
         Task Stop();
     }
@@ -38,6 +41,7 @@ public abstract class BaseNodeHost : INodeHost
         private System.Timers.Timer _gossipTimer;
         private BlockingCollection<IPeerHost> _peers;
         private string _nodeId;
+        private bool _isStarted = false;
 
         public BaseNodeHost(IPeerStore peerStore, IPeerHostFactory peerHostFactory, INodeStateStore nodeStateStore, IClusterStore clusterStore, ILogger<BaseNodeHost> logger, IOptions<ConsensusPeerOptions> options)
         {
@@ -52,6 +56,9 @@ public BaseNodeHost(IPeerStore peerStore, IPeerHostFactory peerHostFactory, INod
         }
 
         public BlockingCollection<IPeerHost> Peers => _peers;
+        public INodeStateStore NodeStateStore => _nodeStateStore;
+        public bool IsStarted => _isStarted;
+        public int Port => _options.Port;
         public bool IsRunning { get; private set; }
         public UdpClient UdpServer { get; private set; }
         public ILogger<BaseNodeHost> Logger { get; private set; }
@@ -82,6 +89,7 @@ public Task Stop()
             IsRunning = false;
             _proxyClient.Close();
             UdpServer.Close();
+            _isStarted = false;
             return Task.CompletedTask;
         }
 
@@ -90,6 +98,7 @@ protected async Task InternalRun()
             try
             {
                 if (NodeStarted != null) NodeStarted(this, new EventArgs());
+                _isStarted = true;
                 while (true)
                 {
                     TokenSource.Token.ThrowIfCancellationRequested();
@@ -187,6 +196,7 @@ private void StopGossipTimer()
         private async Task BroadcastGossipHeartbeat(object source, ElapsedEventArgs e)
         {
             var nodes = await _clusterStore.GetAllNodes(TokenSource.Token);
+            nodes = nodes.Where(n => n.Port != _options.Port || n.Url != _options.Url);
             var totalNodes = nodes.Count();
             var nbNodes = _options.GossipMaxNodeBroadcast;
             if(totalNodes < nbNodes) nbNodes = totalNodes;
@@ -254,10 +264,24 @@ private async Task<GossipPackage> HandleGossipRequest(GossipUpdateNodeStateReque
         private async Task<GossipPackage> HandleGossipRequest(GossipJoinNodeRequest request)
         {
             var cluster = await _clusterStore.GetNode(request.Url, request.Port, TokenSource.Token);
-            if (cluster == null) await _clusterStore.AddNode(new ClusterNode { Port = request.Port, Url = request.Url }, TokenSource.Token);
-            using (var gossipClient = new GossipClient(request.Url, request.Port))
+            if (cluster == null) 
             {
-                // await gossipClient.JoinNode(_options.Url, _options.Port, true, TokenSource.Token);
+                await _clusterStore.AddNode(new ClusterNode { Port = request.Port, Url = request.Url }, TokenSource.Token);
+                var allNodes = await _clusterStore.GetAllNodes(TokenSource.Token);
+                using (var gossipClient = new GossipClient(request.Url, request.Port))
+                {
+                    await gossipClient.UpdateClusterNodes(_options.Url, _options.Port, allNodes.Select(n => new ClusterNodeMessage { Port = n.Port, Url = n.Url }).ToList(), TokenSource.Token);
+                }
+            }
+
+            return null;
+        }
+
+        private async Task<GossipPackage> HandleGossipRequest(GossipUpdateClusterRequest request)
+        {
+            foreach(var node in request.Nodes)
+            {
+                await _clusterStore.AddNode(new ClusterNode { Port = node.Port, Url = node.Url }, TokenSource.Token);
             }
 
             return null;
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/Models/ClusterNode.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/Models/ClusterNode.cs
@@ -1,4 +1,5 @@
 ﻿using FaasNet.RaftConsensus.Client;
+using System;
 using System.Text.Json;
 
 namespace FaasNet.RaftConsensus.Core.Models
@@ -12,7 +13,7 @@ public NodeState ToNodeState()
         {
             return new NodeState
             {
-                EntityId = string.Empty,
+                EntityId = Guid.NewGuid().ToString(),
                 EntityType = StandardEntityTypes.Cluster,
                 EntityVersion = 0,
                 Value = JsonSerializer.Serialize(this)
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/ServiceCollectionExtensions.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/ServiceCollectionExtensions.cs
@@ -14,15 +14,14 @@ public static ServerBuilder AddConsensusPeer(this IServiceCollection services, A
             if (callback != null) services.Configure(callback);
             else services.Configure<ConsensusPeerOptions>((o) => { });
             var peerStore = new InMemoryPeerStore(new ConcurrentBag<PeerInfo>());
-            var nodeStateStore = new InMemoryNodeStateStore(new ConcurrentBag<NodeState>());
             services.AddLogging();
             services.AddTransient<INodeHost, StandaloneNodeHost>();
             services.AddTransient<IPeerHostFactory, PeerHostFactory>();
             services.AddTransient<IClusterStore, ClusterStore>();
             services.AddScoped<IPeerHost, StandalonePeerHost>();
             services.AddScoped<ILogStore, InMemoryLogStore>();
+            services.AddSingleton<INodeStateStore, InMemoryNodeStateStore>();
             services.AddSingleton<IPeerStore>(peerStore);
-            services.AddSingleton<INodeStateStore>(nodeStateStore);
             return new ServerBuilder(services);
         }
     }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/ClusterStore.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/ClusterStore.cs
@@ -19,22 +19,21 @@ public interface IClusterStore
     public class ClusterStore : IClusterStore
     {
         private readonly INodeStateStore _nodeStateStore;
-        private readonly ConsensusPeerOptions _options;
 
-        public ClusterStore(INodeStateStore nodeStateStore, IOptions<ConsensusPeerOptions> options)
+        public ClusterStore(INodeStateStore nodeStateStore)
         {
             _nodeStateStore = nodeStateStore;
-            _options = options.Value;
         }
 
         public async Task<IEnumerable<ClusterNode>> GetAllNodes(CancellationToken cancellationToken)
         {
             var lastEntityTypes = await _nodeStateStore.GetAllLastEntityTypes(StandardEntityTypes.Cluster, cancellationToken);
+            lastEntityTypes = lastEntityTypes.OrderBy(e => e.EntityVersion);
             var result = lastEntityTypes.Select(et => JsonSerializer.Deserialize<ClusterNode>(et.Value, new JsonSerializerOptions
             {
                 PropertyNameCaseInsensitive = true
             }));
-            return result.Where(r => r.Url != _options.Url || r.Port != _options.Port);
+            return result;
         }
 
         public async Task<ClusterNode> GetNode(string url, int port, CancellationToken cancellationToken)
@@ -43,10 +42,12 @@ public async Task<ClusterNode> GetNode(string url, int port, CancellationToken c
             return allNodes.FirstOrDefault(n => n.Port == port && n.Url == url);
         }
 
-        public Task AddNode(ClusterNode node, CancellationToken cancellationToken)
+        public async Task AddNode(ClusterNode node, CancellationToken cancellationToken)
         {
-            _nodeStateStore.Add(node.ToNodeState());
-            return Task.CompletedTask;
+            var lastEntityType = await _nodeStateStore.GetLastEntityType(StandardEntityTypes.Cluster, cancellationToken);
+            var nodeState = node.ToNodeState();
+            if (lastEntityType != null) nodeState.EntityVersion = lastEntityType.EntityVersion + 1;
+            _nodeStateStore.Add(nodeState);
         }
     }
 }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/INodeStateStore.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/Stores/INodeStateStore.cs
@@ -13,6 +13,7 @@ public interface INodeStateStore
         void Add(NodeState nodeState);
         void Update(NodeState nodeState);
         Task<int> SaveChanges(CancellationToken cancellationToken);
+        Task<IEnumerable<NodeState>> GetAllEntityTypes(CancellationToken cancellationToken);
         Task<IEnumerable<NodeState>> GetAllLastEntityTypes(CancellationToken cancellationToken);
         Task<IEnumerable<NodeState>> GetAllLastEntityTypes(string entityType, CancellationToken cancellationToken);
         Task<IEnumerable<NodeState>> GetAllSpecificEntityTypes(List<(string EntityType, int EntityVersion)> parameter, CancellationToken cancellationToken);
@@ -23,6 +24,11 @@ public class InMemoryNodeStateStore : INodeStateStore
     {
         private readonly ConcurrentBag<NodeState> _nodeStates;
 
+        public InMemoryNodeStateStore() 
+        {
+            _nodeStates = new ConcurrentBag<NodeState>();
+        }
+
         public InMemoryNodeStateStore(ConcurrentBag<NodeState> nodeStates)
         {
             _nodeStates = nodeStates;
@@ -38,6 +44,12 @@ public void Update(NodeState nodeState)
             _nodeStates.Remove(nodeState);
         }
 
+        public Task<IEnumerable<NodeState>> GetAllEntityTypes(CancellationToken cancellationToken)
+        {
+            IEnumerable<NodeState> result = _nodeStates.OrderByDescending(ns => ns.EntityVersion);
+            return Task.FromResult(result);
+        }
+
         public Task<IEnumerable<NodeState>> GetAllLastEntityTypes(CancellationToken cancellationToken)
         {
             IEnumerable<NodeState> result = _nodeStates.OrderByDescending(ns => ns.EntityVersion).GroupBy(ns => ns.EntityType).Select(ns => ns.First());
@@ -53,7 +65,7 @@ public Task<IEnumerable<NodeState>> GetAllLastEntityTypes(string entityType, Can
         public Task<IEnumerable<NodeState>> GetAllSpecificEntityTypes(List<(string EntityType, int EntityVersion)> parameter, CancellationToken cancellationToken)
         {
             IEnumerable<NodeState> result = _nodeStates.Where(ns => parameter.Any(p => p.EntityType == ns.EntityType && p.EntityVersion == ns.EntityVersion));
-            throw new System.NotImplementedException();
+            return Task.FromResult(result);
         }
 
         public Task<NodeState> GetLastEntityType(string entityType, CancellationToken cancellationToken)
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs
diff --git a/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs b/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@ public class GossipCommands: BaseCommands`
`8`	`8`	`public static GossipCommands SYNC_RESULT = new GossipCommands(3, "SYNC_RESULT");`
`9`	`9`	`public static GossipCommands UPDATE_NODE_STATE_REQUEST = new GossipCommands(4, "UPDATE_NODE_STATE_REQUEST");`
`10`	`10`	`public static GossipCommands JOIN_NODE_REQUEST = new GossipCommands(5, "JOIN_NODE_REQUEST");`
	`11`	`+ public static GossipCommands UPDATE_CLUSTER_NODES_REQUEST = new GossipCommands(6, "UPDATE_CLUSTER_NODES_REQUEST");`
`11`	`12`
`12`	`13`	`protected GossipCommands(int code)`
`13`	`14`	`{`
Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,13 @@ public static GossipPackage Deserialize(ReadBufferContext context)`
`56`	`56`	`return result;`
`57`	`57`	`}`
`58`	`58`
	`59`	`+ if (header.Command == GossipCommands.UPDATE_CLUSTER_NODES_REQUEST)`
	`60`	`+ {`
	`61`	`+ var result = new GossipUpdateClusterRequest { Header = header };`
	`62`	`+ result.Extract(context);`
	`63`	`+ return result;`
	`64`	`+ }`
	`65`	`+`
`59`	`66`	`return new GossipPackage { Header = header };`
`60`	`67`	`}`
`61`	`68`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,5 +31,14 @@ public static GossipPackage AddNode(string url, int port)`
`31`	`31`	`Url = url`
`32`	`32`	`};`
`33`	`33`	`}`
	`34`	`+`
	`35`	`+ public static GossipPackage UpdateClusterNodes(string url, int port, ICollection<ClusterNodeMessage> clusterNodes)`
	`36`	`+ {`
	`37`	`+ return new GossipUpdateClusterRequest`
	`38`	`+ {`
	`39`	`+ Header = new GossipHeader(GossipCommands.UPDATE_CLUSTER_NODES_REQUEST, url, port),`
	`40`	`+ Nodes = clusterNodes`
	`41`	`+ };`
	`42`	`+ }`
`34`	`43`	`}`
`35`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`using FaasNet.RaftConsensus.Client;`
	`2`	`+using System;`
`2`	`3`	`using System.Text.Json;`
`3`	`4`
`4`	`5`	`namespace FaasNet.RaftConsensus.Core.Models`
`@@ -12,7 +13,7 @@ public NodeState ToNodeState()`
`12`	`13`	`{`
`13`	`14`	`return new NodeState`
`14`	`15`	`{`
`15`		`- EntityId = string.Empty,`
	`16`	`+ EntityId = Guid.NewGuid().ToString(),`
`16`	`17`	`EntityType = StandardEntityTypes.Cluster,`
`17`	`18`	`EntityVersion = 0,`
`18`	`19`	`Value = JsonSerializer.Serialize(this)`
Original file line number	Diff line number	Diff line change
`@@ -19,22 +19,21 @@ public interface IClusterStore`
`19`	`19`	`public class ClusterStore : IClusterStore`
`20`	`20`	`{`
`21`	`21`	`private readonly INodeStateStore _nodeStateStore;`
`22`		`- private readonly ConsensusPeerOptions _options;`
`23`	`22`
`24`		`- public ClusterStore(INodeStateStore nodeStateStore, IOptions<ConsensusPeerOptions> options)`
	`23`	`+ public ClusterStore(INodeStateStore nodeStateStore)`
`25`	`24`	`{`
`26`	`25`	`_nodeStateStore = nodeStateStore;`
`27`		`- _options = options.Value;`
`28`	`26`	`}`
`29`	`27`
`30`	`28`	`public async Task<IEnumerable<ClusterNode>> GetAllNodes(CancellationToken cancellationToken)`
`31`	`29`	`{`
`32`	`30`	`var lastEntityTypes = await _nodeStateStore.GetAllLastEntityTypes(StandardEntityTypes.Cluster, cancellationToken);`
	`31`	`+ lastEntityTypes = lastEntityTypes.OrderBy(e => e.EntityVersion);`
`33`	`32`	`var result = lastEntityTypes.Select(et => JsonSerializer.Deserialize<ClusterNode>(et.Value, new JsonSerializerOptions`
`34`	`33`	`{`
`35`	`34`	`PropertyNameCaseInsensitive = true`
`36`	`35`	`}));`
`37`		`- return result.Where(r => r.Url != _options.Url \|\| r.Port != _options.Port);`
	`36`	`+ return result;`
`38`	`37`	`}`
`39`	`38`
`40`	`39`	`public async Task<ClusterNode> GetNode(string url, int port, CancellationToken cancellationToken)`
`@@ -43,10 +42,12 @@ public async Task<ClusterNode> GetNode(string url, int port, CancellationToken c`
`43`	`42`	`return allNodes.FirstOrDefault(n => n.Port == port && n.Url == url);`
`44`	`43`	`}`
`45`	`44`
`46`		`- public Task AddNode(ClusterNode node, CancellationToken cancellationToken)`
	`45`	`+ public async Task AddNode(ClusterNode node, CancellationToken cancellationToken)`
`47`	`46`	`{`
`48`		`- _nodeStateStore.Add(node.ToNodeState());`
`49`		`- return Task.CompletedTask;`
	`47`	`+ var lastEntityType = await _nodeStateStore.GetLastEntityType(StandardEntityTypes.Cluster, cancellationToken);`
	`48`	`+ var nodeState = node.ToNodeState();`
	`49`	`+ if (lastEntityType != null) nodeState.EntityVersion = lastEntityType.EntityVersion + 1;`
	`50`	`+ _nodeStateStore.Add(nodeState);`
`50`	`51`	`}`
`51`	`52`	`}`
`52`	`53`	`}`