Riak Core on Partisan on Elixir Tutorial: Getting Started

Creating the project

mix new civiledb --app civile
cd civiledb

The project at this stage is available in the tag start-1

defp deps do
  [
    {:uniendo, git: "git://github.com/marianoguerra/uniendo.git", branch: "master"}
  ]
end

def application do
  [
    extra_applications: [:riak_core, :logger],
    mod: {Civile, []}
  ]
end

First build

Let's try building the project as is, mainly to test that the dependencies are compiled correctly.

First we need to get the dependencies:

mix deps.get

If it's the first time it may ask you:

Could not find Hex, which is needed to build dependency :folsom
Shall I install Hex? (if running non-interactively, use "mix local.hex --force") [Yn]

Just hit enter to continue.

When it finishes, try to build the project:

mix compile

The project at this stage is available in the tag start-4

If it's the first time you build a project with dependencies written in erlang it may ask you:

Shall I install rebar3? (if running non-interactively, use "mix local.rebar --force") [Yn]

Just hit enter to continue.

If you get an error compiling you may want to try updating the rebar3 version mix uses with:

mix local.rebar rebar3 ~/bin/rebar3

If setup fails to build, try commenting the line on deps/setup/rebar.config:

{post_hooks, [{compile, "make escriptize"}]}.

Like this:

%{post_hooks, [{compile, "make escriptize"}]}.

Making it start

We need to start our application and riak_core, for that we need some initial setup:

mkdir priv config
cp deps/riak_core/priv/riak_core.schema priv/

The project at this stage is available in the tag start-5

Add the following content to the file in config/config.exs:

use Mix.Config

config :riak_core,
  ring_state_dir: 'ring_data_dir',
  handoff_port: 8099,
  handoff_ip: '127.0.0.1',
  schema_dirs: ['priv']

config :sasl,
  errlog_type: :error

The project at this stage is available in the tag start-6

Edit lib/civile.ex and change its content to:

defmodule Civile do
  use Application
  require Logger

  def start(_type, _args) do
    case Civile.Supervisor.start_link do
      {:ok, pid} ->
        {:ok, pid}
      {:error, reason} ->
        Logger.error("Unable to start Civile supervisor because: #{inspect reason}")
    end
  end
end

Create a new file called lib/civile_supervisor.ex with the following content:

defmodule Civile.Supervisor do
  use Supervisor

  def start_link do
    # riak_core appends _sup to the application name.
    Supervisor.start_link(__MODULE__, [], [name: :civile_sup])
  end

  def init(_args) do
    children = []
    supervise(children, strategy: :one_for_one, max_restarts: 5, max_seconds: 10)
  end

end

The project at this stage is available in the tag start-7

Now recompile the project:

mix compile

And start it:

iex --name dev@127.0.0.1 -S mix run

You should see something like this:

Erlang/OTP 22 [erts-10.5] [source] [64-bit] [smp:4:4] [ds:4:4:10] [async-threads:1] [hipe]

00:48:07.530 [info] Starting reporters with []
00:48:07.577 [info] Using node name: 'dev@127.0.0.1'
00:48:07.581 [info] Resolving "127.0.0.1"...
00:48:07.583 [info] Resolved "dev@127.0.0.1" to {127,0,0,1}
00:48:07.583 [info] Resolved "127.0.0.1" to {127,0,0,1}
00:48:07.679 [info] Partisan listening on {127,0,0,1}:49489 listen_addrs: [#{ip => {127,0,0,1},port => 49489}]
00:48:07.686 [info] Not using container orchestration; disabling.
00:48:07.694 [info] node 'dev@127.0.0.1' choosing random seed: {52969312,-576460751045187665,-576460752303423453}
00:48:07.728 [info] node 'dev@127.0.0.1' choosing random seed: {52969312,-576460751045187665,-576460752303423453}
00:48:07.774 [info] Configuring partisan dispatch: false
dets: file "data/cluster_meta/manifest.dets" not properly closed, repairing ...
00:48:07.938 [info] New capability: {riak_core,vnode_routing} = proxy
00:48:07.942 [info] New capability: {riak_core,staged_joins} = true
00:48:07.947 [info] New capability: {riak_core,resizable_ring} = true
00:48:07.951 [info] New capability: {riak_core,fold_req_version} = v2
00:48:07.956 [info] New capability: {riak_core,security} = true
00:48:07.960 [info] New capability: {riak_core,bucket_types} = true
00:48:07.964 [info] New capability: {riak_core,net_ticktime} = true
Interactive Elixir (1.9.1) - press Ctrl+C to exit (type h() ENTER for help)
iex(dev@127.0.0.1)1>

You may get a lager crash log at startup, ignore it.

Hit Ctrl+C twice to quit.

Implementing our VNode

Write the following on a new file called lib/civile_vnode.ex:

defmodule Civile.VNode do
  @behaviour :riak_core_vnode

  def start_vnode(partition) do
    :riak_core_vnode_master.get_vnode_pid(partition, __MODULE__)
  end

  def init([partition]) do
    {:ok, %{partition: partition}}
  end

  def handle_command({:ping, v}, _sender, state = %{partition: partition}) do
     {:reply, {:pong, v + 1, node(), partition}, state}
  end

  def handoff_starting(_dest, state) do
    {true, state}
  end

  def handoff_cancelled(state) do
    {:ok, state}
  end

  def handoff_finished(_dest, state) do
    {:ok, state}
  end

  def handle_handoff_command(_fold_req, _sender, state) do
    {:noreply, state}
  end

  def is_empty(state) do
    {true, state}
  end

  def terminate(_reason, _state) do
    :ok
  end

  def delete(state) do
    {:ok, state}
  end

  def handle_handoff_data(_bin_data, state) do
    {:reply, :ok, state}
  end

  def encode_handoff_item(_k, _v) do
  end

  def handle_coverage(_req, _key_spaces, _sender, state) do
    {:stop, :not_implemented, state}
  end

  def handle_exit(_pid, _reason, state) do
    {:noreply, state}
  end

  def handle_overload_command(_, _, _) do
    :ok
  end

  def handle_overload_info(_, _idx) do
    :ok
  end
end

Write the following on a new file called lib/civile_service.ex:

defmodule Civile.Service do

  def ping(v\\1) do
    idx = :riak_core_util.chash_key({"civile", "ping#{v}"})
    pref_list = :riak_core_apl.get_primary_apl(idx, 1, Civile.Service)

    [{index_node, _type}] = pref_list

    :riak_core_vnode_master.sync_command(index_node, {:ping, v}, Civile.VNode_master)
  end

end

In lib/civile_supervisor.ex, add the vnode master as a child to our supervisor

Change:

def init(_args) do
  children = []
  supervise(children, strategy: :one_for_one, max_restarts: 5, max_seconds: 10)
end

To:

def init(_args) do
  children = [
    worker(:riak_core_vnode_master, [Civile.VNode], id: Civile.VNode_master_worker)
  ]
  supervise(children, strategy: :one_for_one, max_restarts: 5, max_seconds: 10)
end

And register our vnode implementation on riak_core.

In lib/civile.ex change:

def start(_type, _args) do
  case Civile.Supervisor.start_link do
    {:ok, pid} ->
      {:ok, pid}
    {:error, reason} ->
      Logger.error("Unable to start Civile supervisor because: #{inspect reason}")
  end
end

To:

def start(_type, _args) do
  case Civile.Supervisor.start_link do
    {:ok, pid} ->
      :ok = :riak_core.register(vnode_module: Civile.VNode)
      :ok = :riak_core_node_watcher.service_up(Civile.Service, self())
      {:ok, pid}
    {:error, reason} ->
      Logger.error("Unable to start Civile supervisor because: #{inspect reason}")
  end
end

The project at this stage is available in the tag start-8

Now compile and run again:

mix compile
iex --name dev@127.0.0.1 -S mix run

Inside the shell try our new service:

iex(dev@127.0.0.1)1> Civile.Service.ping
{:pong, 2, :"dev@127.0.0.1",
 251195593916248939066258330623111144003363405824}

iex(dev@127.0.0.1)2> Civile.Service.ping 32
{:pong, 33, :"dev@127.0.0.1",
 342539446249430371453988632667878832731859189760}

The response is a tuple that contains the atom :pong, the number we passed (or 1 by default) incremented by one, the node and the partition that handled the reply. Right now we only have one node so that's not useful, but in the next steps it will make more sense.

Playing with clustering (in the same machine)

To build a cluster on the same machine we will need to make 3 builds with slightly different configurations to avoid each running node from reading another node's files or trying to use another node's ports, for this we will create 3 different configs and make 3 different builds.

First add a line at the end of config/config.exs:

import_config "#{Mix.env}.exs"

This will import the configuration for the given mix environment

Now create the following files:

use Mix.Config

config :riak_core,
  node: 'dev1@127.0.0.1',
  web_port: 8198,
  handoff_port: 8199,
  ring_state_dir: 'ring_data_dir_1',
  platform_data_dir: 'data_1'

use Mix.Config

config :riak_core,
  node: 'dev2@127.0.0.1',
  web_port: 8298,
  handoff_port: 8299,
  ring_state_dir: 'ring_data_dir_2',
  platform_data_dir: 'data_2'

use Mix.Config

config :riak_core,
  node: 'dev3@127.0.0.1',
  web_port: 8398,
  handoff_port: 8399,
  ring_state_dir: 'ring_data_dir_3',
  platform_data_dir: 'data_3'

use Mix.Config

MIX_ENV=dev1 iex --name dev1@127.0.0.1 -S mix run
MIX_ENV=dev2 iex --name dev2@127.0.0.1 -S mix run
MIX_ENV=dev3 iex --name dev3@127.0.0.1 -S mix run

:riak_core.join('dev1@127.0.0.1')

{:ok, ring} = :riak_core_ring_manager.get_my_ring
:riak_core_ring.pretty_print(ring, [:legend])

Civile.Service.ping 32

iex(dev1@127.0.0.1)2> Civile.Service.ping 32
{:pong, 33, :"dev3@127.0.0.1",
 342539446249430371453988632667878832731859189760}

iex(dev2@127.0.0.1)5> Civile.Service.ping 32
{:pong, 33, :"dev3@127.0.0.1",
 342539446249430371453988632667878832731859189760}

iex(dev3@127.0.0.1)12> Civile.Service.ping 32
{:pong, 33, :"dev3@127.0.0.1",
 342539446249430371453988632667878832731859189760}