Impl leo-project/leofs#530

test/leo_object_storage_api_tests.erl

@@ -151,6 +151,32 @@ compaction_1_test_() ->
        {timeout, 1000, fun compact_1/0}}
      ]}.
 
+compaction_3_test_() ->
+    {setup,
+     fun ( ) ->
+             ?debugVal("### COMPACTION.START #3 ###"),
+             os:cmd("rm -rf " ++ ?AVS_DIR_FOR_COMPACTION),
+             application:start(sasl),
+             application:start(os_mon),
+             application:start(crypto),
+             application:start(leo_object_storage),
+             ok
+     end,
+     fun (_) ->
+             application:stop(crypto),
+             application:stop(os_mon),
+             application:stop(sasl),
+             timer:sleep(1000),
+             application:stop(leo_object_storage),
+             timer:sleep(1000),
+             ?debugVal("### COMPACTION.END #3 ###"),
+             ok
+     end,
+     [
+      {"test compaction",
+       {timeout, 1000, fun compact_3/0}}
+     ]}.
+
 diagnose() ->
     %% Launch object-storage
     leo_object_storage_api:start([{1, ?AVS_DIR_FOR_COMPACTION}]),
@@ -346,12 +372,78 @@ compact_1() ->
     ?assertEqual(TotalNum, ActiveNum),
     ok.
 
+compact_3() ->
+    %% Launch object-storage
+    leo_object_storage_api:start([{4, ?AVS_DIR_FOR_COMPACTION}]),
+
+    %% inserted but will be ignored due to the key size limit
+    Key = list_to_binary(lists:duplicate(4100, $X)),
+    CMeta = [{?PROP_CMETA_CLUSTER_ID, 'leofs_1'},
+             {?PROP_CMETA_NUM_OF_REPLICAS, 3},
+             {?PROP_CMETA_VER, leo_date:clock()},
+             {?PROP_CMETA_UDM, [{<<"name">>, <<"LeoFS">>},
+                                {<<"category">>, <<"distributed storage">>},
+                                {<<"url">>, <<"leo-project.net/leofs">>}
+                               ]
+             }],
+    put_irregular_bin(Key, CMeta, 0, leo_date:clock()),
+
+    Expected = lists:duplicate(10, ok),
+    Actual = [ put_regular_bin(1, 50) || _ <- lists:seq(1, 10) ],
+    ?assertEqual(Expected, Actual),
+    Min = 1024 * 1024,
+    Range = Min * 2,
+    ok = put_irregular_bin(Min, Range),
+
+    %% Execute compaction
+    timer:sleep(3000),
+    {ok, Stats} = leo_object_storage_api:stats(),
+    {SumTotal0, SumActive0} =
+                 lists:foldl(
+                   fun(#storage_stats{total_num  = Total,
+                                      active_num = Active},
+                       {SumTotal, SumActive}) ->
+                           {SumTotal + Total, SumActive + Active}
+                   end, {0, 0}, Stats),
+    ?assertEqual(501, SumTotal0),
+    ?assertEqual(51, SumActive0),
+    FunHasChargeOfNode = fun(_Key_,_NumOfReplicas_) ->
+                                 true
+                         end,
+    TargetPids = [_Pid || {_Pid,_}
+                              <- leo_object_storage_api:get_object_storage_pid(all)],
+    ok = leo_compact_fsm_controller:run(TargetPids, 3, FunHasChargeOfNode),
+
+    %% Check comaction status
+    ok = check_status(),
+
+    %% Check # of active objects and total of objects
+    timer:sleep(1000),
+    {ok, Stats1} = leo_object_storage_api:stats(),
+    {SumTotal1, SumActive1} =
+                 lists:foldl(
+                   fun(#storage_stats{total_num  = Total,
+                                      active_num = Active},
+                       {SumTotal, SumActive}) ->
+                           {SumTotal + Total, SumActive + Active}
+                   end, {0, 0}, Stats1),
+    ?assertEqual(50, SumTotal1),
+    ?assertEqual(50, SumActive1),
+    ok.
+
 
 check_status() ->
     timer:sleep(100),
     ID = 'leo_compact_worker_0',
+    %% shouldn't be blocked
+    {_, _} = sys:get_state(ID),
     case leo_compact_fsm_controller:state() of
         {ok, #compaction_stats{status = ?ST_IDLING}} ->
+            %% check pid_pairs is empty
+            {_, State} = sys:get_state(leo_compact_fsm_controller),
+            ?debugVal(State),
+            %% 1:record name, 2: id, 3: server_pairs, 4: pid_pairs
+            ?assertEqual([], erlang:element(4, State)),
             ok;
         {ok, #compaction_stats{locked_targets = []}} ->
             check_status();
@@ -465,7 +557,11 @@ put_regular_bin_with_cmeta(Index, Counter) ->
 %% @private
 put_irregular_bin() ->
     Min = 1024 * 16,
-    Len = case erlang:phash2(leo_date:clock(), (1024 * 512)) of
+    Range = 1024 * 512,
+    put_irregular_bin(Min, Range).
+
+put_irregular_bin(Min, Range) ->
+    Len = case erlang:phash2(leo_date:clock(), Range) of
               Val when Val < Min ->
                   Min;
               Val ->
@@ -478,6 +574,25 @@ put_irregular_bin() ->
     io:format(user, "[garbage] *** start:~p, end:~p~n", [Offset - byte_size(Bin), Offset]),
     ok.
 
+put_irregular_bin(Key, CMeta, CSize, Clock) ->
+    AddrId = 1,
+    Bin = crypto:rand_bytes(erlang:phash2(leo_date:clock(), (1024 * 1024))),
+    CMetaBin = leo_object_storage_transformer:list_to_cmeta_bin(CMeta),
+
+    Object = #?OBJECT{method    = put,
+                      addr_id   = AddrId,
+                      key       = Key,
+                      ksize     = byte_size(Key),
+                      data      = Bin,
+                      dsize     = byte_size(Bin),
+                      csize     = CSize,
+                      meta      = CMetaBin,
+                      msize     = byte_size(CMetaBin),
+                      checksum  = leo_hex:raw_binary_to_integer(crypto:hash(md5, Bin)),
+                      timestamp = leo_date:now(),
+                      clock     = Clock
+                     },
+    leo_object_storage_api:put({AddrId, Key}, Object).
 
 %% @private
 delete_metadata(0) ->