<commit_before>%% ===================================================================
%% @author <NAME>
%% @copyright (C): 2014 ACK CYFRONET AGH
%% This software is released under the MIT license
%% cited in 'LICENSE.txt'.
%% @end
%% ===================================================================
%% @doc: This test checks correctness and amount of write operations
%% using FUSE.
%% @end
%% ===================================================================

-module(veilcluster_driver_dd).
-export([setup/0, new/1, run/4]).

-include("basho_bench.hrl").
-include("registered_names.hrl").
-include("communication_protocol_pb.hrl").
-include("veil_modules/dao/dao.hrl").
-include("veil_modules/fslogic/fslogic.hrl").

%% ====================================================================
%% Test driver callbacks
%% ====================================================================

%% setup/0
%% ====================================================================
%% @doc Runs once per each test node at begging of a test (before any new/1 is called)
-spec setup() -> no_return().
%% ====================================================================
setup() ->
    try
        ?INFO("Test setup~n", []),

        case basho_bench_config:get(client_id) of
            1 -> setup_storages(); %% If its the first test node, initialize cluster
            _ -> timer:sleep(2000) %% Otherwise wait for main node to finish
        %% TODO: implement better, more deterministic way of synchronising test nodes (e.g. via ready-ping)
        end
    catch
        E1:E2 -> ?ERROR("Setup error: ~p:~p~n", [E1, E2])
    end.


%% new/1
%% ====================================================================
%% @doc Creates new worker with integer id
-spec new(Id :: integer()) -> Result when
    Result :: {ok, term()} | {error, Reason :: term()}.
%% ====================================================================
new(Id) ->
    try
        ?INFO("Initializing worker with id: ~p~n", [Id]),
        VFSRoot = basho_bench_config:get(veilfs_root),
        Dir = VFSRoot ++ "/stress_test_" ++ basho_bench_config:get(build_id),
        File = Dir ++ "/file_" ++ integer_to_list(Id),

        case make_dir(Dir) of
            ok ->
                %% Open test file (each test process gets different file name like "file_Id")
                Device = case open_helper(File, {error, first_try}, 20) of
                             {error, Reason} ->
                                 ?ERROR("Can not open file: ~p: ~p", [File, Reason]),
                                 Reason;
                             IO -> IO
                         end,
                BlockSize = basho_bench_config:get(block_size),
                Data = [0 || _X <- lists:seq(1, 1024 * BlockSize)],
                Args = {Device, 0, Data},
                ?INFO("Worker with id: ~p initialized successfully with arguments: ~p", [Id, Args]),
                {ok, {Device, 0, Data}};
            {error, Reason} ->
                ?ERROR("Can not create directory: ~p: ~p~n", [Dir, Reason]),
                {error, Reason}
        end
    catch
        E1:E2 ->
            ?ERROR("Initialization error for worker with id: ~p: ~p", [Id, {E1, E2}]),
            {error, {E1, E2}}
    end.


%% run/4
%% ====================================================================
%% @doc Runs an operation using one of workers
%% Currently only 'write' operation is implemented
-spec run(Operation :: atom(), KeyGen :: fun(), ValueGen :: fun(), State :: term()) -> Result when
    Result :: {ok, NewState :: term()} | {error, Reason :: term(), NewState :: term()}.
%% ====================================================================
run(write, _KeyGen, _ValueGen, {Dev, _Offset, _Data} = State) when is_atom(Dev) ->
    timer:sleep(1000), %% Dont generate more then one error per sec when open/2 is failing
    {error, {open, Dev}, State};
run(write, _KeyGen, _ValueGen, {Dev, Offset, Data}) ->
    NewState = {Dev, (Offset + length(Data)) rem (basho_bench_config:get(max_filesize) * 1024 * 1024), Data},
    case file:pwrite(Dev, 0, Data) of
        ok -> {ok, NewState};
        {error, Reason} ->
            ?ERROR("Run error: (file: ~p, offset: ~p): ~p", [Dev, Offset, Reason]),
            {error, Reason, NewState}
    end;
run(Operation, _KeyGen, _ValueGen, State) ->
    ?ERROR("Unknown operation ~p in state ~p", [Operation, State]),
    {error, unknown_operation, State}.


%% ====================================================================
%% Helper functions
%% ====================================================================

%% setup_storages/0
%% ====================================================================
%% @doc Register test user and configure storages on cluster
-spec setup_storages() -> no_return().
%% ====================================================================
setup_storages() ->
    ?INFO("Storage setup~n", []),
    CertFile = basho_bench_config:get(cert_file),
    ?INFO("CertFile: ~p~n", [CertFile]),
    [Host | _] = basho_bench_config:get(cluster_hosts),
    ?INFO("Host: ~p~n", [Host]),
    Groups = #fuse_group_info{name = ?CLUSTER_FUSE_ID, storage_helper = #storage_helper_info{name = "DirectIO", init_args = ["/mnt/gluster"]}},
    Worker = map_hostname(Host),
    ?INFO("Worker: ~p~n", [Worker]),
    {ok, DN} = get_dn(CertFile),
    ?INFO("DN: ~p~n", [DN]),

    %% Init net kernet in order to connect to cluster
    Net = net_kernel:start([list_to_atom("tester@" ++ net_adm:localhost()), longnames]),
    ?INFO("Net: ~p~n", [Net]),
    Cookie = erlang:set_cookie(node(), veil_cluster_node),
    ?INFO("Cookie: ~p~n", [Cookie]),

    case rpc:call(Worker, os, cmd, ["rm -rf /mnt/gluster/*"]) of
        "" -> ?INFO("RM OK~n", []);
        RmError -> throw(io_lib:fwrite("Can not remove files from storage: ~p", [RmError]))
    end,

    case rpc:call(Worker, fslogic_storage, insert_storage, ["ClusterProxy", [], [Groups]]) of
        {ok, _} -> ?INFO("INSERT OK~n", []);
        InsertError -> throw(io_lib:fwrite("Can not insert storage: ~p", [InsertError]))
    end,

    case rpc:call(Worker, user_logic, create_user, ["veilfstestuser", "Test Name", [], "<EMAIL>", [DN]]) of
        {ok, _} -> ?INFO("CREATE OK~n", []);
        CreateError -> throw(io_lib:fwrite("Can not add test user: ~p", [CreateError]))
    end.


%% get_dn/1
%% ====================================================================
%% @doc Gets rDN list compatibile user_logic:create_user from PEM file
-spec get_dn(PEMFile :: string()) -> Result when
    Result :: {ok, DN :: string()}.
%% ====================================================================
get_dn(PEMFile) -><commit_msg>VFS-692 Remove logs.<commit_after>%% ===================================================================
%% @author <NAME>
%% @copyright (C): 2014 ACK CYFRONET AGH
%% This software is released under the MIT license
%% cited in 'LICENSE.txt'.
%% @end
%% ===================================================================
%% @doc: This test checks correctness and amount of write operations
%% using FUSE.
%% @end
%% ===================================================================

-module(veilcluster_driver_dd).
-export([setup/0, new/1, run/4]).

-include("basho_bench.hrl").
-include("registered_names.hrl").
-include("communication_protocol_pb.hrl").
-include("veil_modules/dao/dao.hrl").
-include("veil_modules/fslogic/fslogic.hrl").

%% ====================================================================
%% Test driver callbacks
%% ====================================================================

%% setup/0
%% ====================================================================
%% @doc Runs once per each test node at begging of a test (before any new/1 is called)
-spec setup() -> no_return().
%% ====================================================================
setup() ->
    try
        ?INFO("Test setup~n", []),

        case basho_bench_config:get(client_id) of
            1 -> setup_storages(); %% If its the first test node, initialize cluster
            _ -> timer:sleep(2000) %% Otherwise wait for main node to finish
        %% TODO: implement better, more deterministic way of synchronising test nodes (e.g. via ready-ping)
        end
    catch
        E1:E2 -> ?ERROR("Setup error: ~p:~p~n", [E1, E2])
    end.


%% new/1
%% ====================================================================
%% @doc Creates new worker with integer id
-spec new(Id :: integer()) -> Result when
    Result :: {ok, term()} | {error, Reason :: term()}.
%% ====================================================================
new(Id) ->
    try
        ?INFO("Initializing worker with id: ~p~n", [Id]),
        VFSRoot = basho_bench_config:get(veilfs_root),
        Dir = VFSRoot ++ "/stress_test_" ++ basho_bench_config:get(build_id),
        File = Dir ++ "/file_" ++ integer_to_list(Id),

        case make_dir(Dir) of
            ok ->
                %% Open test file (each test process gets different file name like "file_Id")
                Device = case open_helper(File, {error, first_try}, 20) of
                             {error, Reason} ->
                                 ?ERROR("Can not open file: ~p: ~p", [File, Reason]),
                                 Reason;
                             IO -> IO
                         end,
                BlockSize = basho_bench_config:get(block_size),
                Data = [0 || _X <- lists:seq(1, 1024 * BlockSize)],
                Args = {Device, 0, Data},
                ?INFO("Worker with id: ~p initialized successfully with arguments: ~p", [Id, Args]),
                {ok, {Device, 0, Data}};
            {error, Reason} ->
                ?ERROR("Can not create directory: ~p: ~p~n", [Dir, Reason]),
                {error, Reason}
        end
    catch
        E1:E2 ->
            ?ERROR("Initialization error for worker with id: ~p: ~p", [Id, {E1, E2}]),
            {error, {E1, E2}}
    end.


%% run/4
%% ====================================================================
%% @doc Runs an operation using one of workers
%% Currently only 'write' operation is implemented
-spec run(Operation :: atom(), KeyGen :: fun(), ValueGen :: fun(), State :: term()) -> Result when
    Result :: {ok, NewState :: term()} | {error, Reason :: term(), NewState :: term()}.
%% ====================================================================
run(write, _KeyGen, _ValueGen, {Dev, _Offset, _Data} = State) when is_atom(Dev) ->
    timer:sleep(1000), %% Dont generate more then one error per sec when open/2 is failing
    {error, {open, Dev}, State};
run(write, _KeyGen, _ValueGen, {Dev, Offset, Data}) ->
    NewState = {Dev, (Offset + length(Data)) rem (basho_bench_config:get(max_filesize) * 1024 * 1024), Data},
    case file:pwrite(Dev, 0, Data) of
        ok -> {ok, NewState};
        {error, Reason} ->
            ?ERROR("Run error: (file: ~p, offset: ~p): ~p", [Dev, Offset, Reason]),
            {error, Reason, NewState}
    end;
run(Operation, _KeyGen, _ValueGen, State) ->
    ?ERROR("Unknown operation ~p in state ~p", [Operation, State]),
    {error, unknown_operation, State}.


%% ====================================================================
%% Helper functions
%% ====================================================================

%% setup_storages/0
%% ====================================================================
%% @doc Register test user and configure storages on cluster
-spec setup_storages() -> no_return().
%% ====================================================================
setup_storages() ->
    ?INFO("Storage setup~n", []),
    CertFile = basho_bench_config:get(cert_file),
    [Host | _] = basho_bench_config:get(cluster_hosts),
    Groups = #fuse_group_info{name = ?CLUSTER_FUSE_ID, storage_helper = #storage_helper_info{name = "DirectIO", init_args = ["/mnt/gluster"]}},
    Worker = map_hostname(Host),
    {ok, DN} = get_dn(CertFile),

    %% Init net kernet in order to connect to cluster
    case net_kernel:start([list_to_atom("tester@" ++ net_adm:localhost()), longnames]) of
        {ok, _} -> ok;
        NetError -> throw(io_lib:fwrite("Can not start net_kernel: ~p", [NetError]))
    end,

    erlang:set_cookie(node(), veil_cluster_node),

    case rpc:call(Worker, os, cmd, ["rm -rf /mnt/gluster/*"]) of
        "" -> ok;
        RmError -> throw(io_lib:fwrite("Can not remove files from storage: ~p", [RmError]))
    end,

    case rpc:call(Worker, fslogic_storage, insert_storage, ["ClusterProxy", [], [Groups]]) of
        {ok, _} -> ok;
        InsertError -> throw(io_lib:fwrite("Can not insert storage: ~p", [InsertError]))
    end,

    case rpc:call(Worker, user_logic, create_user, ["veilfstestuser", "<NAME>", [], "<EMAIL>", [DN]]) of
        {ok, _} -> ok;
        CreateError -> throw(io_lib:fwrite("Can not add test user: ~p", [CreateError]))
    end.


%% get_dn/1
%% ====================================================================
%% @doc Gets rDN list compatibile user_logic:create_user from PEM file
-spec get_dn(PEMFile :: string()) -> Result when
    Result :: {ok, DN :: string()}.
%% ====================================================================
get_dn(PEMFile) ->
{'fraction_non_alphanumeric': 0.3104929895974672, 'fraction_numerical': 0.008065731946328961, 'mean_word_length': 3.5528483184625945, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 66, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '963642', 'n_tokens_mistral': 3923, 'n_tokens_neox': 3602, 'n_words': 1008}