Example raft cluster consensus

[LCastilloSymbiotic]

Hi,

I read your documentation and try to enter in this web link: https://github.com/sakno/dotNext/tree/develop/src/examples/RaftNode , but is show 404 error. I would like to know if you have some simple example project to use consensus for some data.

Greetings,

Luis

[sakno]

Hi @LCastilloSymbiotic , the link is broken because the project has moved to .NET Foundation. The correct link is https://github.com/dotnet/dotNext/tree/develop/src/examples/RaftNode

[LCastilloSymbiotic]

Hi @sakno , I try to replicate your example, but when i try to import IClusterMemberLifetime from DotNext.Net.Cluster.Consensus.Raft; it's show a not found exeception. Do know you why?

[sakno]

Could you please provide a full stack trace?

[LCastilloSymbiotic]

I have this code, that's the same of your example:

using DotNext.Net.Cluster;
using DotNext.Net.Cluster.Consensus.Raft;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;

namespace Services.Replication
{
    internal sealed class ClusterConfigurator : IClusterMemberLifetime
    {
        internal static void LeaderChanged(ICluster cluster, IClusterMember? leader)
        {
            Debug.Assert(cluster is IRaftCluster);
            var term = ((IRaftCluster)cluster).Term;
            var timeout = ((IRaftCluster)cluster).ElectionTimeout;
            Console.WriteLine(leader is null
                ? "Consensus cannot be reached"
                : $"New cluster leader is elected. Leader address is {leader.EndPoint}");
            Console.WriteLine($"Term of local cluster member is {term}. Election timeout {timeout}");
        }

        public void Initialize(IRaftCluster cluster, IDictionary<string, string> metadata)
        {
            cluster.LeaderChanged += LeaderChanged;
        }

        public void Shutdown(IRaftCluster cluster)
        {
            cluster.LeaderChanged -= LeaderChanged;
        }
    }
}

Generate this error:

Severity Code Description Project File Line Suppression State
Error CS0246 The type or namespace name 'IClusterMemberLifetime' could not be found (are you missing a using directive or an assembly reference?) "" C:............\Replication\ClusterConfigurator.cs 11 Active

[sakno]

Check your ProjectReference/PackageReference elements in csproj file and import paths. The interface is definitely public and even visible on FuGet package explorer: https://www.fuget.org/packages/DotNext.AspNetCore.Cluster/3.3.0/lib/netcoreapp3.1/DotNext.AspNetCore.Cluster.dll/DotNext.Net.Cluster.Consensus.Raft/IClusterMemberLifetime

[LCastilloSymbiotic]

Yes, I check, but i think the PakageReference is ok:

[sakno]

IClusterMemberLifetime interface is in DotNext.AspNetCore.Cluster package

[LCastilloSymbiotic]

In your example you import DotNext.Net.Cluster:

[sakno]

No, it has both imports: https://github.com/dotnet/dotNext/blob/master/src/examples/RaftNode/RaftNode.csproj#L25

[LCastilloSymbiotic]

Ok, that's work!!!

[LCastilloSymbiotic]

Hi, when execute this code:

 app.UseConsensusProtocolHandler()
                .RedirectToLeader(LeaderResource)
                .UseRouting()
                .UseEndpoints(endpoints =>
                {
                    endpoints.MapGet(LeaderResource, RedirectToLeader);
                });

raise this error:

No service for type 'DotNext.Net.Cluster.Consensus.Raft.Http.Embedding.RaftEmbeddedCluster' has been registered.',

Do you know why?

This is the complete code of the startup:

 public class Startup
    {
        private readonly IConfiguration configuration;

        public Startup(IConfiguration configuration)
        {
            this.configuration = configuration;
            Newtonsoft.Json.JsonConvert.DefaultSettings = () => new CustomJsonSettings();
        }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            // Configuration of NewtonsoftJson
            services.AddControllers()
                .AddNewtonsoftJson();
            services.AddMvc()
                .SetCompatibilityVersion(CompatibilityVersion.Latest)
                .AddJsonOptions(options =>
                    options.JsonSerializerOptions.MaxDepth = 10)
                .AddNewtonsoftJson(options => 
                    options.SerializerSettings.MaxDepth = 10)
                .AddJsonOptions(options =>
                    options.JsonSerializerOptions.MaxDepth = 10);
            Newtonsoft.Json.JsonConvert.DefaultSettings = () => 
                new Newtonsoft.Json.JsonSerializerSettings() {
                    MaxDepth = 10
                };


            // Configuration of Cluster
            services.ConfigureCluster<ClusterConfigurator>()
               .AddSingleton<IHttpMessageHandlerFactory, RaftClientHandlerFactory>()
               .AddOptions()
               .AddRouting();

            var path = configuration[SimplePersistentState.LogLocation];
            if (!string.IsNullOrWhiteSpace(path))
            {
                services.AddSingleton<AppEventSource>();
                services.UsePersistenceEngine<IValueProvider, SimplePersistentState>()
                    .AddSingleton<IHostedService, DataModifier>();
            }
        }



        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            // app.UseHttpsRedirection();

            app.UseRouting();
            app.UseAuthorization();
            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });


            // Configuration of Cluster
            const string LeaderResource = "/leader";


            // Check
            app.UseConsensusProtocolHandler()
                .RedirectToLeader(LeaderResource)
                .UseRouting()
                .UseEndpoints(endpoints =>
                {
                    endpoints.MapGet(LeaderResource, RedirectToLeader);
                });
        }

        private static Task RedirectToLeader(HttpContext context)
        {
            var cluster = context.RequestServices.GetRequiredService<IRaftCluster>();
            return context.Response.WriteAsync($"Leader address is {cluster.Leader?.EndPoint}. Current address is {context.Connection.LocalIpAddress}:{context.Connection.LocalPort}", context.RequestAborted);
        }
    }

[sakno]

Probably you forgot to call JoinCluster method on WebHost. I recommend you to start from the following article: https://dotnet.github.io/dotNext/features/cluster/aspnetcore.html. There is a good explanation of differences between Embedded and Hosted modes.

[LCastilloSymbiotic]

Hi, not i call this, you can see the code of the Program:

    public class Program
    {
        public static void Main(string[] args)
        {
            var config = new ConfigurationBuilder()
                .AddJsonFile("appsettings.json", optional: false)
                .Build();

            _ = RaftManager.Main(config);
            CreateHostBuilder(args).Build().Run();
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureWebHostDefaults(webBuilder =>
                {
                    webBuilder.UseStartup<Startup>();
                });
    }

and the code for the RaftManager:

public static class RaftManager
    {
        //private RaftCluster cluster = null;
        
        //public RaftManager(string[][] clusterPeers, int localPort)
        //{

        //    RaftCluster.NodeConfiguration config = 
        //        new RaftCluster.TcpConfiguration(new IPEndPoint(IPAddress.Loopback, localPort));

        //    // Cluster configuration
        //    config.Partitioning = true;


        //    for (int i = 0; i < clusterPeers.Length; i++)
        //    {
        //        config.Members.Add(new IPEndPoint(
        //            System.Net.IPAddress.Parse(clusterPeers[i][0]),
        //            Int32.Parse(clusterPeers[i][1])));
        //    }

        //    cluster = new RaftCluster(config);

        //}

        private static Task UseAspNetCoreHost(int port, string? persistentStorage = null)
        {
            var configuration = new Dictionary<string, string>
            {
                {"partitioning", "false"},
                {"lowerElectionTimeout", "150" },
                {"upperElectionTimeout", "300" },
                {"members:0", "https://localhost:63755"},
                {"requestJournal:memoryLimit", "5" },
                {"requestJournal:expiration", "00:01:00" }
            };
            if (!string.IsNullOrEmpty(persistentStorage))
                configuration[SimplePersistentState.LogLocation] = persistentStorage;
            return new HostBuilder().ConfigureWebHost(webHost =>
            {
                webHost.UseKestrel(options =>
                {
                    options.ListenLocalhost(port, listener => listener.UseHttps(LoadCertificate()));
                })
                .UseStartup<Startup>();
            })
            .ConfigureLogging(builder => builder.AddConsole().SetMinimumLevel(LogLevel.Error))
            .ConfigureAppConfiguration(builder => builder.AddInMemoryCollection(configuration))
            .JoinCluster()
            .Build()
            .RunAsync();
        }

        private static async Task UseConfiguration(RaftCluster.NodeConfiguration config, string? persistentStorage)
        {
            config.Members.Add(new IPEndPoint(IPAddress.Loopback, 63755));
            var loggerFactory = new LoggerFactory();
            var loggerOptions = new ConsoleLoggerOptions
            {
                LogToStandardErrorThreshold = LogLevel.Warning
            };
            loggerFactory.AddProvider(new ConsoleLoggerProvider(new FakeOptionsMonitor<ConsoleLoggerOptions>(loggerOptions)));
            config.LoggerFactory = loggerFactory;

            using var cluster = new RaftCluster(config);
            cluster.LeaderChanged += ClusterConfigurator.LeaderChanged;
            var modifier = default(DataModifier?);
            if (!string.IsNullOrEmpty(persistentStorage))
            {
                var state = new SimplePersistentState(persistentStorage, new AppEventSource());
                cluster.AuditTrail = state;
                modifier = new DataModifier(cluster, state);
            }
            await cluster.StartAsync(CancellationToken.None);
            await (modifier?.StartAsync(CancellationToken.None) ?? Task.CompletedTask);
            using var handler = new CancelKeyPressHandler();
            Console.CancelKeyPress += handler.Handler;
            await handler.WaitAsync();
            Console.CancelKeyPress -= handler.Handler;
            await (modifier?.StopAsync(CancellationToken.None) ?? Task.CompletedTask);
            await cluster.StopAsync(CancellationToken.None);
        }

        private static Task UseUdpTransport(int port, string? persistentStorage)
        {
            var configuration = new RaftCluster.UdpConfiguration(new IPEndPoint(IPAddress.Loopback, port))
            {
                LowerElectionTimeout = 150,
                UpperElectionTimeout = 300,
                DatagramSize = 1024
            };
            return UseConfiguration(configuration, persistentStorage);
        }

        private static Task UseTcpTransport(int port, string? persistentStorage, bool useSsl)
        {
            var configuration = new RaftCluster.TcpConfiguration(new IPEndPoint(IPAddress.Loopback, port))
            {
                LowerElectionTimeout = 150,
                UpperElectionTimeout = 300,
                TransmissionBlockSize = 4096,
                SslOptions = useSsl ? CreateSslOptions() : null
            };

            return UseConfiguration(configuration, persistentStorage);

            static SslOptions CreateSslOptions()
            {
                var options = new SslOptions();
                options.ServerOptions.ServerCertificate = LoadCertificate();
                options.ClientOptions.TargetHost = "localhost";
                options.ClientOptions.RemoteCertificateValidationCallback = RaftClientHandlerFactory.AllowCertificate;
                return options;
            }
        }

        private static Task StartNode(string protocol, int port, string? persistentStorage = null)
        {
            switch (protocol.ToLowerInvariant())
            {
                case "http":
                case "https":
                    return UseAspNetCoreHost(port, persistentStorage);
                case "udp":
                    return UseUdpTransport(port, persistentStorage);
                case "tcp":
                    return UseTcpTransport(port, persistentStorage, false);
                case "tcp+ssl":
                    return UseTcpTransport(port, persistentStorage, true);
                default:
                    Console.Error.WriteLine("Unsupported protocol type");
                    Environment.ExitCode = 1;
                    return Task.CompletedTask;
            }
        }

        public static async Task Main(IConfiguration configuration)
        {
            string protocol = configuration["ReplicationConfig:Protocol"];
            int port = int.Parse(configuration["ReplicationConfig:LocalPort"]);
            string persistentStorage = (configuration["ReplicationConfig:PersistentStorage"] == "") ? null : configuration["ReplicationConfig:PersistentStorage"];

            switch (persistentStorage)
            {
                case null:
                    await StartNode(protocol, port);
                    break;
                default:
                    await StartNode(protocol, port, persistentStorage);
                    break;
            }
        }

        private static X509Certificate2 LoadCertificate()
        {
            var certificate = new X509Certificate2("C:\\Users\\User\\Documents\\Symbiotic\\Git\\ToP-PMS\\PaymentManagementServer\\PaymentManagementServer.Services\\Certificates\\node.pfx",
                "1234", 
                X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
            return certificate;
        }
    }

And the parameters are:

PersistentStorage: null
Protocol: "http"
LocalPort: 63754