Merge pull request #594 from basho/feature/bitcask-tombstone2-upgrade

Verify Bitcask tombstone 2 upgrade

src/rt.erl

@@ -81,6 +81,7 @@
          members_according_to/1,
          owners_according_to/1,
          partition/2,
+         partitions_for_node/1,
          pbc/1,
          pbc_read/3,
          pbc_read/4,
@@ -916,6 +917,11 @@ check_singleton_node(Node) ->
     ?assertEqual([Node], Owners),
     ok.
 
+% @doc Get list of partitions owned by node (primary).
+partitions_for_node(Node) ->
+    Ring = get_ring(Node),
+    [Idx || {Idx, Owner} <- riak_core_ring:all_owners(Ring), Owner == Node].
+
 %% @doc Get the raw ring for `Node'.
 get_ring(Node) ->
     {ok, Ring} = rpc:call(Node, riak_core_ring_manager, get_raw_ring, []),

tests/verify_bitcask_tombstone2_upgrade.erl

@@ -0,0 +1,91 @@
+% @doc Verify that upgrading Riak with Bitcask to 2.0 or later will trigger
+% an upgrade mechanism that will end up merging all existing bitcask files.
+% This is necessary so that old style tombstones are reaped, which might
+% otherwise stay around for a very long time. This version writes tombstones
+% that can be safely dropped during a merge. Bitcask could resurrect old 
+% values easily when reaping tombstones during a partial merge if a
+% restart happened later.
+-module(verify_bitcask_tombstone2_upgrade).
+-behaviour(riak_test).
+-export([confirm/0]).
+-include_lib("eunit/include/eunit.hrl").
+
+confirm() ->
+    TestMetaData = riak_test_runner:metadata(),
+    Backend = proplists:get_value(backend, TestMetaData),
+    lager:info("Running with backend (this better be Bitcask!) ~p", [Backend]),
+    ?assertEqual({backend, bitcask}, {backend, Backend}),
+    OldVsn = proplists:get_value(upgrade_version, TestMetaData, previous),
+    % Configure for fast merge checks
+    Config = [{riak_kv, [{bitcask_merge_check_interval, 2000}]},
+              {bitcask, [{max_file_size, 100}]}],
+    Nodes = rt:build_cluster([{OldVsn, Config}]),
+    verify_bitcask_tombstone2_upgrade(Nodes),
+    pass.
+
+% Expects nodes running a version of Riak < 2.0 using Bitcask
+verify_bitcask_tombstone2_upgrade(Nodes) ->
+    lager:info("Write some data, write it good"),
+    write_some_data(Nodes),
+    lager:info("Collect the list of bitcask files created"),
+    BitcaskFiles = list_bitcask_files(Nodes),
+    lager:info("Now update the node to the current version"),
+    [rt:upgrade(Node, current) || Node <- Nodes],
+    lager:info("And wait until all the old files have been merged, the version upgrade finished"),
+    ?assertEqual(ok, rt:wait_until(upgrade_complete_fun(BitcaskFiles))),
+    lager:info("And that is that").
+
+write_some_data([Node1 | _]) ->
+    rt:pbc_systest_write(Node1, 10000).
+
+list_bitcask_files(Nodes) ->
+    [{Node, list_node_bitcask_files(Node)} || Node <- Nodes].
+
+list_node_bitcask_files(Node) ->
+    % Gather partitions owned, list *.bitcask.data on each.
+    Partitions = rt:partitions_for_node(Node),
+    {ok, DataDir} = rt:rpc_get_env(Node, [{bitcask, data_root}]),
+    [begin
+         IdxStr = integer_to_list(Idx),
+         IdxDir = filename:join(DataDir, IdxStr),
+         BitcaskPattern = filename:join([IdxDir, "*.bitcask.data"]),
+         Paths = rpc:call(Node, filelib, wildcard, [BitcaskPattern]),
+         ?assert(is_list(Paths)),
+         Files = [filename:basename(Path) || Path <- Paths],
+         {IdxDir, Files}
+     end || Idx <- Partitions].
+
+upgrade_complete_fun(BitcaskFiles) ->
+    fun() ->
+            upgrade_complete(BitcaskFiles)
+    end.
+
+upgrade_complete(BitcaskFiles) ->
+    all(true, [upgrade_complete(Node, PFiles)
+               || {Node, PFiles} <- BitcaskFiles]).
+
+upgrade_complete(Node, PartitionFiles) ->
+    all(true,[upgrade_complete(Node, IdxDir, Files)
+              || {IdxDir, Files} <- PartitionFiles]).
+
+upgrade_complete(Node, IdxDir, Files) ->
+    % Check we have version.txt, no upgrade.txt, no merge.txt
+    MergeFile = filename:join(IdxDir, "merge.txt"),
+    UpgradeFile = filename:join(IdxDir, "upgrade.txt"),
+    VsnFile = filename:join(IdxDir, "version.txt"),
+    file_exists(Node, VsnFile) andalso
+        not file_exists(Node, UpgradeFile) andalso
+        not file_exists(Node, MergeFile) andalso
+        all(false,
+            [file_exists(Node, filename:join(IdxDir, F)) || F <- Files]).
+
+file_exists(Node, Path) ->
+    case rpc:call(Node, filelib, is_regular, [Path]) of
+        {badrpc, Reason} ->
+            throw({can_not_check_file, Node, Path, Reason});
+        Result ->
+            Result
+    end.
+
+all(Val, L) ->
+    lists:all(fun(E) -> E == Val end, L).