Weekend project means just that, I’m trying some things out, and writing something real is the best way to exercise. This isn’t going to be a full blown project, but it should be functional and usable.

The basic idea, I’m going to build a distributed key/value configuration store. Similar to etcd, this will allow me to explore how to handle full blown Rachis from both server & client sides.

We want this to be a full  blown implementation, which means persistence, snapshots, network api, the works.

In terms of the data model, we’ll go for the simplest possible one. A key/value store. A key is a string of up to 128 characters. A value is a json formatted value of up to 16Kb. Persistence will be handled by Voron. The persistent of the project is mostly Voron, so what we are left with is the following:

public enum KeyValueOperationTypes
{
    Add,
    Del
}

public class KeyValueOperation
{
    public KeyValueOperationTypes Type;
    public string Key;
    public JToken Value;
}

public class OperationBatchCommand : Command
{
    public KeyValueOperation[] Batch { get; set; }
}

This gives us the background for the actual state machine:

public class KeyValueStateMachine : IRaftStateMachine
{
    readonly StorageEnvironment _storageEnvironment;

    public KeyValueStateMachine(StorageEnvironmentOptions options)
    {
        _storageEnvironment = new StorageEnvironment(options);
        using (var tx = _storageEnvironment.NewTransaction(TransactionFlags.ReadWrite))
        {
            _storageEnvironment.CreateTree(tx, "items");
            var metadata = _storageEnvironment.CreateTree(tx, "$metadata");
            var readResult = metadata.Read("last-index");
            if (readResult != null)
                LastAppliedIndex = readResult.Reader.ReadLittleEndianInt64();
            tx.Commit();
        }
    }

    public event EventHandler<KeyValueOperation> OperatonExecuted;

    protected void OnOperatonExecuted(KeyValueOperation e)
    {
        var handler = OperatonExecuted;
        if (handler != null) handler(this, e);
    }

    public JToken Read(string key)
    {
        using (var tx = _storageEnvironment.NewTransaction(TransactionFlags.Read))
        {
            var items = tx.ReadTree("items");

            var readResult = items.Read(key);

            if (readResult == null)
                return null;


            return JToken.ReadFrom(new JsonTextReader(new StreamReader(readResult.Reader.AsStream())));
        }
    }

    public long LastAppliedIndex { get; private set; }

    public void Apply(LogEntry entry, Command cmd)
    {
        var batch = (OperationBatchCommand)cmd;
        Apply(batch.Batch, cmd.AssignedIndex);
    }

    
    private void Apply(IEnumerable<KeyValueOperation> ops, long commandIndex)
    {
        using (var tx = _storageEnvironment.NewTransaction(TransactionFlags.ReadWrite))
        {
            var items = tx.ReadTree("items");
            var metadata = tx.ReadTree("$metadata");
            metadata.Add("last-index", EndianBitConverter.Little.GetBytes(commandIndex));
            var ms = new MemoryStream();
            foreach (var op in ops)
            {
                switch (op.Type)
                {
                    case KeyValueOperationTypes.Add:
                        ms.SetLength(0);

                        var streamWriter = new StreamWriter(ms);
                        op.Value.WriteTo(new JsonTextWriter(streamWriter));
                        streamWriter.Flush();

                        ms.Position = 0;
                        items.Add(op.Key, ms);
                        break;
                    case KeyValueOperationTypes.Del:
                        items.Delete(op.Key);
                        break;
                    default:
                        throw new ArgumentOutOfRangeException();
                }
                OnOperatonExecuted(op);
            }

            tx.Commit();
        }
    }


    public void Dispose()
    {
        if (_storageEnvironment != null)
            _storageEnvironment.Dispose();
    }
}

As you can see, there isn’t much here. Not surprising, since we are storing a key/value data structure. I’m also ignoring snapshots for now. That is good enough for now, let us go for the network portion of the work. We are going to be using Web API for the network stuff. And we’ll be initializing it like so:

var nodeName = options.NodeName ?? (Environment.MachineName + ":" + options.Port);

var kvso = StorageEnvironmentOptions.ForPath(Path.Combine(options.DataPath, "KeyValue"));
using (var statemachine = new KeyValueStateMachine(kvso))
{
    using (var raftEngine = new RaftEngine(new RaftEngineOptions(
        new NodeConnectionInfo
        {
            Name = nodeName,
            Url = new Uri("http://" + Environment.MachineName + ":" + options.Port),
        },
        StorageEnvironmentOptions.ForPath(Path.Combine(options.DataPath, "Raft")),
        new HttpTransport(nodeName),
        statemachine
        )))
    {
        using (WebApp.Start(new StartOptions
        {
            Urls = { "http://+:" + options.Port + "/" }
        }, builder =>
        {
            var httpConfiguration = new HttpConfiguration();
            RaftWebApiConfig.Register(httpConfiguration);
            httpConfiguration.Properties[typeof(HttpTransportBus)] = new HttpTransportBus(nodeName);
            httpConfiguration.Properties[typeof(RaftEngine)] = raftEngine;
            builder.UseWebApi(httpConfiguration);
        }))
        {

            Console.WriteLine("Ready & processing requests, press ENTER to sop");

            Console.ReadLine();
        }
    }
}

Note that we need to initialize both the state machine and the raft engine, then wire the raft engine controllers. Now we are pretty much done with setup, and we can turn to the actual semantics of running the cluster. The first thing that I want to do is to setup the baseline, so we create this base controller:

public abstract class TailFeatherController : ApiController
{
    public KeyValueStateMachine StateMachine { get; private set; }
    public RaftEngine RaftEngine { get; private set; }

    public override async Task<HttpResponseMessage> ExecuteAsync(HttpControllerContext controllerContext, CancellationToken cancellationToken)
    {
        RaftEngine = (RaftEngine)controllerContext.Configuration.Properties[typeof(RaftEngine)];
        StateMachine = (KeyValueStateMachine)RaftEngine.StateMachine;
        try
        {
            return await base.ExecuteAsync(controllerContext, cancellationToken);
        }
        catch (NotLeadingException)
        {
            var currentLeader = RaftEngine.CurrentLeader;
            if (currentLeader == null)
            {
                return new HttpResponseMessage(HttpStatusCode.PreconditionFailed)
                {
                    Content = new StringContent("{ 'Error': 'No current leader, try again later' }")
                };
            }
            var leaderNode = RaftEngine.CurrentTopology.GetNodeByName(currentLeader);
            if (leaderNode == null)
            {
                return new HttpResponseMessage(HttpStatusCode.InternalServerError)
                {
                    Content = new StringContent("{ 'Error': 'Current leader " + currentLeader + " is not found in the topology. This should not happen.' }")
                };
            }
            return new HttpResponseMessage(HttpStatusCode.Redirect)
            {
                Headers =
                {
                    Location = leaderNode.Uri
                }
            };
        }
    }
}

That is a lot of error handling, but basically it just get the right values from the configuration and expose them to the controller actions, then a lot of error handling when we have a command that requires a leader that hit a follower.

Next step, actually managing the cluster, here we go:

public class AdminController : TailFeatherController
{
    [HttpGet]
    [Route("tailfeather/admin/fly-with-us")]
    public async Task<HttpResponseMessage> Join([FromUri] string url, [FromUri] string name)
    {
        var uri = new Uri(url);
        name = name ?? uri.Host + (uri.IsDefaultPort ? "" : ":" + uri.Port);

        await RaftEngine.AddToClusterAsync(new NodeConnectionInfo
        {
            Name = name,
            Uri = uri
        });
        return new HttpResponseMessage(HttpStatusCode.Accepted);
    }

    [HttpGet]
    [Route("tailfeather/admin/fly-away")]
    public async Task<HttpResponseMessage> Leave([FromUri] string name)
    {
        await RaftEngine.RemoveFromClusterAsync(new NodeConnectionInfo
        {
            Name = name
        });
        return new HttpResponseMessage(HttpStatusCode.Accepted);
    }
}

So now we have a way to add and remove items from the cluster, which is all the admin stuff that we need to handle right now. Next, we need to actually wire the operations, this is done here:

public class KeyValueController : TailFeatherController
{
    [HttpGet]
    [Route("tailfeather/key-val/read")]
    public HttpResponseMessage Read([FromUri] string key)
    {
        var read = StateMachine.Read(key);
        if (read == null)
        {
            return Request.CreateResponse(HttpStatusCode.NotFound, new
            {
                RaftEngine.State,
                Key = key,
                Missing = true
            });
        }
        return Request.CreateResponse(HttpStatusCode.OK, new
        {
            RaftEngine.State,
            Key = key,
            Value = read
        });
    }

    [HttpGet]
    [Route("tailfeather/key-val/set")]
    public Task<HttpResponseMessage> Set([FromUri] string key, [FromUri] string val)
    {
        JToken jVal;
        try
        {
            jVal = JToken.Parse(val);
        }
        catch (JsonReaderException)
        {
            jVal = val;
        }

        var op = new KeyValueOperation
        {
            Key = key,
            Type = KeyValueOperationTypes.Add,
            Value = jVal
        };

        return Batch(new[] { op });
    }

    [HttpGet]
    [Route("tailfeather/key-val/del")]
    public Task<HttpResponseMessage> Del([FromUri] string key)
    {
        var op = new KeyValueOperation
        {
            Key = key,
            Type = KeyValueOperationTypes.Del,
        };

        return Batch(new[] { op });
    }

    [HttpPost]
    [Route("tailfeather/key-val/batch")]
    public async Task<HttpResponseMessage> Batch()
    {
        var stream = await Request.Content.ReadAsStreamAsync();
        var operations = new JsonSerializer().Deserialize<KeyValueOperation[]>(new JsonTextReader(new StreamReader(stream)));

        return await Batch(operations);
    }

    private async Task<HttpResponseMessage> Batch(KeyValueOperation[] operations)
    {
        var taskCompletionSource = new TaskCompletionSource<object>();
        RaftEngine.AppendCommand(new OperationBatchCommand
        {
            Batch = operations,
            Completion = taskCompletionSource
        });
        await taskCompletionSource.Task;

        return Request.CreateResponse(HttpStatusCode.Accepted);
    }
}

And we are pretty much set.

Note that I’ve been writing this post while I’m writing the code, so I’ve made some small changes, you can see actual code here.

Anyway, we are pretty much done. Now we can compile and try testing what is going on.

First, we seed the cluster, but running:

.\TailFeather.exe --port=9079 --DataPath=One --Name=One –Bootstrap

This tell us that this node is allowed to become a leader without having to pre-configure a cluster. This command runs and exit, so now we’ll run three such copies:

• start .\TailFeather.exe "--port=9079 --DataPath=One --Name=One"
• start .\TailFeather.exe "--port=9078 --DataPath=Two --Name=Two"
• start .\TailFeather.exe "--port=9077 --DataPath=Three --Name=Three"

We have all three nodes up and running, so now is the time to actually make use of it:

http://localhost:9079/tailfeather/key-val/set?key=ravendb&val={ 'Url': 'http://live-test.ravendb.net', 'Database': 'Sample' }

In this case, you can see that we are setting a configuration value to point to a RavenDB database on the first node. Note that at this point, we have a single node cluster, and the two other are waiting to join it, but are taking no action.

We can get the value back using:

So far, so good. Now, let us add a second node in by inviting it to fly with our cluster. We do that using the following command:

http://localhost:9079/tailfeather/admin/fly-with-us?url=http://localhost:9078&name=Two

Which will give us:

Note that we are using the just added node for too look at this.

Next, we can add the third node.

http://localhost:9079/tailfeather/admin/fly-with-us?url=http://localhost:9077&name=Three

I would put the image in, but I think you get the point.

This is it for now. We have a highly available persistent & distributed key/value store. Next, we need to tackle the idea of snapshots and the client API, but I’ll deal with that at another post.