Verify Bitcask tombstone 2 upgrade

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.

src/rt.erl

@@ -80,6 +80,7 @@
          members_according_to/1,
          owners_according_to/1,
          partition/2,
+         partitions_for_node/1,
          pbc/1,
          pbc_read/3,
          pbc_read/4,
@@ -895,6 +896,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).