This evenings entertainment is
“Once a job is done, the result should be stored in the file layer, together with the logs.“
A new test needs to be devised to prove the result is stored in the file layer once a job is done. Of course the test is put in wn_job_layer_tests.erl . The test can be seen below – and as always – the test is written first.
stored_in_file_layer() -> %% (1) wn_resource_layer:register(#wn_resource{name = "Laptop", type = [{laptop,1}], resides = node() }), %% (2) Path = create_file_at(?NODE_ROOT), %% (3) File1 = #wn_file{id = "File1",file = Path,resides = node()}, Job1 = #wn_job{id = "JobId", files = [File1], resources = [laptop], commands = ["cat EunitFile","touch MadeFile"], timeout = 100 }, %% (4) ok = wn_job_layer:register(Job1), %% (5) timer:sleep(500), %% (6) {ok,T} = wn_job_layer:finished_at("JobId"), %% (7) TimeSuffix = wn_util:time_marker_to_string(T), ExpectedFileName = "JobId_result_"++TimeSuffix++".tgz", ExpectedFileId = "JobId_result", ExpectedFiles = [F1,F2,F3] = ["EUnitFile","MadeFile","Log.txt"], %% (8) {ok,ResultId} = wn_job_layer:stored_result("JobId"), ?assertEqual(ExpectedFileId,ResultId), %% (9) [_,Res] = wn_file_layer:list_files(), ?assertEqual(ExpectedFileName,filename:basename(Res#wn_file.file)), ?assertEqual(ExpectedFileId,Res#wn_file.id), %% (10) {ok,Tar} = wn_file_layer:retrieve_file(node(),ResultId), ?assertMatch({ok,ExpectedFiles},erl_tar:table(Tar,[compressed])), %% (11) erl_tar:extract(Tar,[{files,[F3]},compressed]), {ok,IoDev} = file:open("local_stream.txt",[write]), ok = wn_job_layer:stream(IoDev,"JobId"), timer:sleep(100), file:close(IoDev), {ok,LocalStreamBin} = file:read_file("local_stream.txt"), {ok,JobLogBin} = file:read_file(F3), ?assertEqual(LocalStreamBin,JobLogBin), %% (12) erl_tar:extract(Tar,[{files,[F1]},compressed]), {ok,LocalEunitBin} = file:read_file(Path), {ok,JobEunitBin} = file:read_file(F1), ?assertEqual(LocalEunitBin,JobEunitBin), %% (13) erl_tar:extract(Tar,[{files,[F2]},compressed]), ?assertEqual({ok,<<>>},file:read_file(F2)), %% (14) ok = file:delete("local_stream.txt"), ok = file:delete(F1), ok = file:delete(F2), ok = file:delete(F3), ok = file:delete(Tar).
This test is one of the bigger ones, and deserves some guidance. From the top to bottom: (1) resource registration, (2) creation of the file I wish to send as part of the job, (3) file and job declaration, (4) job registration, (5) waiting for the job to finish (one of the structural weaknesses of this test) and (6) job finish checking by getting the timestamp at the moment of finish. (7) Setting up the expected result values for comparison, (8) getting the result-id and comparing it with the expected result. (9) Listing all files in the file layer and comparing with expected result (note that there SHOULD be two files in the layer – [1] First EunitFile, [2] the result package). (10) Retrieving the expected tgz file and checking that it should contain only two specific files (the log and the sent file). (11) Extraction of the payload logfile and checking with a streamed version that it’s the same. (12) Extraction of the job file to see that it’s there, and comparison to see it’s the same content. (13) Testing that the created file is the one which we sent back, (14) Cleanup of the extracted files and the tgz result.
So, of to write the code – first the wn_util:time_marker_to_string/1 function. One important aspect is that this utility function should not be allowed to exists for the sole purpose of the testing – however as we shall see, this function does have other applications and thus survives.
In wn_util.erl, (skipping to show the export, that part should be bread and butter by now)
-spec(time_marker_to_string(time_marker()) -> string()). time_marker_to_string({Date,Time,Now}) -> F = fun(X) -> string:join([integer_to_list(Y)||Y<-tuple_to_list(X)],"_") end, F(Date)++"_"++F(Time)++"_"++F(Now).
following the test-code, the expected filename is already given in (7), and it’s time to write up the wn_job_layer:stored_result/1 function, taking the JobId as parameter and giving the file-name.
Next the missing part in (8), in wn_job_layer.erl , the API function
-spec(stored_result(string()) -> {ok,string()} | {error,term()}). stored_result(Id) -> gen_server:call(?MODULE,{stored_result,Id}).
and internal logic for handling it
handle_call({stored_result,Id},_,State) -> Result = try_get_stored(Id,State), {reply,Result,State};
and the seen helper function for stored result retrieval.
try_get_stored(Id,State) -> case ets:lookup(State#state.jobs,Id) of [{Id,JobKeeperPid,_}] -> wn_job_keeper:get_stored_result(JobKeeperPid); [] -> {error,no_such_job} end.
This directly leads to the need in fleshing out the wn_job_keeper code. Thus, time to write the wn_job_keeper:get_stored_result/1 function. However, as you are not inside my head (thanks for that!) I might need to give a quick explanation. The idea is that the job keeper holds the name of the stored result, not the result itself. So it’s not the actual tgz we get here. Remember that all files should be handled through the wn_file_layer.
In wn_job_keeper.erl: In order to maintain the stored result in the state, a new field has to be added (and while doing so I took the liberty of making the state type-declarations nicer)
-record(state, {job :: #wn_job{}, info :: [{time(),term()}], result :: [{time(),term()}], stored_result :: string(), stream :: [pid()], done_at :: undefined | time_marker() }).
onward and downwards (in the source file), the API function
-spec(get_stored_result(pid()) -> {ok,string()} | {error,term()}). get_stored_result(Pid) -> gen_fsm:sync_send_all_state_event(Pid,get_stored_result).
alongside the internal function for handling the call
handle_sync_event(get_stored_result,_,X,State) -> {reply, case State#state.stored_result of undefined -> {error,no_result}; R -> {ok,R} end,X,State};
while working my way through this I noticed a bug in the get_done clause of handle_sync_event with done as StateName, and fixed it (should now be what is seen below)
handle_sync_event(get_done,_From,done,State) -> {reply,{ok,State#state.done_at},done,State};
The natural question is now – who called some API function to put the value we are retrieving? The wn_job_worker would seem to be a good place to do this from since the job_worker knows when it’s finished and hence can generate the resulting file, put it into the file layer and signal the wn_job_keeper.
So, going to wn_job_worker.erl, to the part when all commands have been executed, (the handle_info clause of the gen_server) – this is what I put there (not working – needs to be implemented)
handle_info({Port,eof},#state{port = Port} = State) -> case State#state.commands of [] -> wn_job_keeper:info(State#state.pid,no_more_commands), wn_job_keeper:info(State#state.pid,building_result_tgz), ok = file:set_cwd(State#state.olddir), Keeper = State#state.pid, TimeMarker = wn_util:time_marker(), wn_job_keeper:done(Keeper,TimeMarker), Job = wn_job_keeper:get_job(Keeper), Logs = wn_job_keeper:logs(Keeper), {Id,Name} = make_tgz_result(TimeMarker, State#state.workdir, Job,Logs), ok = wn_file_layer:add_file(#wn_file{id = Id, file = Name, resides = node()}), file:delete(Name), wn_job_keeper:set_stored_result(Keeper,Id), {stop,normal,State}; [C|Commands] -> wn_job_keeper:info(State#state.pid,{executing,C}), NewPort = erlang:open_port({spawn,C},[eof,{line,2048}]), {noreply,State#state{commands = Commands,port = NewPort}} end;
This block of code jumps out of the current worker-dir, creates a resulting tgz, adds all files in the worker-dir to it, signals done to the wn_job_keeper and puts the tgz file into the file layer. Of course this is designing as we go and hence there are some functions missing. Finishing off the stuff here first, the local make_tgz_result/4 function
make_tgz_result(TimeMarker,Dir,WnJob,Logs) -> Id = WnJob#wn_job.id++"_result", Name = Id++"_"++wn_util:time_marker_to_string(TimeMarker)++".tgz", {ok,TarFile} = erl_tar:open(Name,[write,compressed]), Files = filelib:wildcard(Dir++"*"), lists:foreach( fun(File) -> ok = erl_tar:add(TarFile,File,filename:basename(File),[]) end,Files), LogStr = lists:foldl( fun({TimeMark,Entry},Str) -> Str++io_lib:format("~p : ~p~n",[TimeMark,Entry]) end,"",lists:sort(lists:append([Lines || {_,Lines} <- Logs]))), ok = erl_tar:add(TarFile,erlang:list_to_binary(LogStr),"Log.txt",[]), erl_tar:close(TarFile), {Id,Name}.
luckily this one does not rely on any more internal / external functions. So that closes this branch of the dev. Backtracking to the handle_info clause, there are some new external functions which has to be developed, one of them is the wn_util:time_marker/0
function.
In wn_util.erl i add the following
-spec(time_marker() -> time_marker()). time_marker() -> {date(),time(),now()}.
the way of signalling that a job is done has now changed, and the second argument is the time_marker of when this happened – this as the wn_job_worker must create the resulting tgz file with the correct time-stamp in the filename.
So, to change the done-signaling, the new lines in wn_job_keeper.erl are
-spec(done(pid(),time_marker()) -> ok). done(Pid,TimeMarker) -> gen_fsm:send_all_state_event(Pid,{done,TimeMarker}).
the internal clause for the handle_event to process this change is
handle_event({done,TimeMarker}, working, #state{info = Info} = State) -> Now = now(), stream_msg(State#state.stream,{Now,done}), {next_state, done, State#state{info = [{Now,done}|Info], done_at = TimeMarker }};
Next – still handle_info of wn_job_worker.erl there is a new function to be implemented, wn_job_keeper:logs/1 which will retrieve the stdout result of the processing to be put into the Log.txt file that was seen in make_tgz_result/4.
Going into wn_job_keeper.erl , i add the following type-spec and function
-spec(logs(pid()) -> [{info|result,[{time(),term()}]}]). logs(Pid) -> gen_fsm:sync_send_all_state_event(Pid,get_logs).
and the function clause to handle the log retrieval request
handle_sync_event(get_logs,_From,X,State) -> {reply,[{info,State#state.info}, {result,State#state.result}],X,State};
Going back to the handle_info clause in wn_job_worker.erl, the next new thing to implement is the function for setting the result inside the wn_job_keeper. The implementation is thus naturally put in wn_job_keeper.erl.
-spec(set_stored_result(pid(),string()) -> ok). set_stored_result(Pid,Id) -> gen_fsm:sync_send_all_state_event(Pid,{stored_result,Id}).
Of course we also need the clause to handle this synchronous request
handle_sync_event({stored_result,Id},_,X,State) -> {reply,ok,X,State#state{stored_result = Id}};
very straight forward and nothing fishy on that sandwich. Ending the wn_job_worker’s handle_info clause, some changes where made in the wn_resource_process.erl. The wn_resource_process should now no longer signal done to the wn_job_keeper as the wn_job_worker does it herself.
As the right of done-signalling has been revoked from the wn_resource_process, some code modification had to be done to the handle_info clause in wn_resource_process.erl
handle_info({'EXIT',WorkerPid,_}, State) -> {noreply, begin [{WorkerPid,_,QueueType}] = ets:lookup(State#state.working,WorkerPid), true = ets:delete(State#state.working,WorkerPid), case lists:keytake(QueueType,1,State#state.queues) of {value,{_,[]},_} -> case ets:lookup(State#state.slots,QueueType) of [{QueueType,infinity}] -> ignore; [{QueueType,X}] -> ets:insert(State#state.slots,{QueueType,X+1}) end, State; {value,{Type,[QueuedPid|R]},Queues} -> case try_dispatch_job(QueuedPid,State,QueueType) of ok -> State#state{queues = [{Type,R}|Queues]}; {error,taken} -> State end end end}.
As the wn_job_worker:make_tgz_result/4 dictated, the input in the Log.txt file changed what the streamer should get – in unison with (11) in the test. Thus this format change has to be accommodated in the wn_job_layer:stream/2 function
-spec(stream(term(),string()) -> ok | {error,term()}). stream(IoDev,Id) -> Stream = fun(F) -> receive {T,E} -> io:format(IoDev,"~p : ~p~n",[T,E]) end, F(F) end, StreamPid = spawn_link(fun() -> Stream(Stream) end), case gen_server:call(?MODULE,{stream,StreamPid,Id}) of ok -> ok; Err -> exit(StreamPid,Err), Err end.
Also, the wn_job_layer:replay/2 function had to be changed
replay(Pid,#state{info = Info,result=Result}) -> lists:foreach(fun(Entry) -> Pid ! Entry end, lists:sort(Info++Result)).
Finally! This satisfies the test! However, there is one point of concern. One of the previous tests can cause this one to fail! The culprit is
{"Queueus on resource type amount", fun queues_on_resource_types_amount/0},
That culprit-test queued several jobs but never removed them properly – causing some interesting effects on the current working directory of the processes. The effect of having this intermediate test interfering is a good example of bad test behaviour. In the best case – we would be expecting each test to be set up in it’s own clean universe, and to die in that same universe as well, without overspilling into other parallel realities!
As can be seen – the make test shows the failure
1> ======================== EUnit ======================== module 'wn_job_layer' module 'wn_job_layer_tests' wn_job_layer_tests: local_test_ (Can register locally)...[0.004 s] ok wn_job_layer_tests: local_test_ (Executed locally)...{1298,929708,203063} : file_fetching_done {1298,929708,203067} : executing_commands {1298,929708,203217} : {executing,"more EUnitFile"} {1298,929708,210033} : "1,2,3" {1298,929708,210134} : no_more_commands {1298,929708,210137} : building_result_tgz {1298,929708,210187} : done [1.004 s] ok wn_job_layer_tests: local_test_ (Executed queue)...{1298,929709,215132} : file_fetching_done {1298,929709,215134} : executing_commands {1298,929709,215221} : {executing,"file EunitFile"} {1298,929709,222661} : "EunitFile: ASCII text" {1298,929709,222721} : no_more_commands {1298,929709,222725} : building_result_tgz {1298,929709,222801} : done {1298,929709,225999} : file_fetching_done {1298,929709,226001} : executing_commands {1298,929709,226189} : {executing,"cat EUnitFile"} {1298,929709,231033} : "1,2,3" {1298,929709,231087} : no_more_commands {1298,929709,231090} : building_result_tgz {1298,929709,231149} : done [1.104 s] ok wn_job_layer_tests: local_test_ (Queueus on resource type amount)...[0.001 s] ok shell-init: error retrieving current directory: getcwd: cannot access parent directories: No such file or directory wn_job_layer_tests: local_test_ (Canceled in queue)...ok wn_job_layer_tests: local_test_ (Done Job Stored in file layer)... =ERROR REPORT==== 28-Feb-2011::21:48:30 === ** Generic server <0.174.0> terminating ** Last message in was {'$gen_cast',{signal,<0.178.0>,a}} ** When Server state == {state,"/Users/zenon/ErlangBlog/worker_net-0.1/node_root/", [{a,[]}], 127001,122904} ** Reason for termination == ** {{badmatch, {error, {function_clause, [{wn_job_worker,'-init/1-fun-0-',[{error,enoent}]}, {wn_job_worker,init,1}, {gen_server,init_it,6}, {proc_lib,init_p_do_apply,3}]}}}, [{wn_resource_process,try_dispatch_job,3}, {wn_resource_process,handle_cast,2}, {gen_server,handle_msg,5}, {proc_lib,init_p_do_apply,3}]} *failed* ::error:{badmatch,{error,not_done}} in function wn_job_layer_tests:stored_in_file_layer/0 [done in 2.687 s] [done in 2.687 s] ======================================================= Failed: 1. Skipped: 0. Passed: 5. zen:worker_net-0.1 zenon$
While the version in where we comment out the test in question succeeds quite nice
1> ======================== EUnit ======================== module 'wn_job_layer' module 'wn_job_layer_tests' wn_job_layer_tests: local_test_ (Can register locally)...[0.005 s] ok wn_job_layer_tests: local_test_ (Executed locally)...{1298,929858,474814} : file_fetching_done {1298,929858,474817} : executing_commands {1298,929858,474949} : {executing,"more EUnitFile"} {1298,929858,481366} : "1,2,3" {1298,929858,481451} : no_more_commands {1298,929858,481454} : building_result_tgz {1298,929858,481489} : done [1.005 s] ok wn_job_layer_tests: local_test_ (Executed queue)...{1298,929859,486690} : file_fetching_done {1298,929859,486693} : executing_commands {1298,929859,486733} : {executing,"file EunitFile"} {1298,929859,493820} : "EunitFile: ASCII text" {1298,929859,493937} : no_more_commands {1298,929859,493939} : building_result_tgz {1298,929859,494065} : done {1298,929859,497623} : file_fetching_done {1298,929859,497627} : executing_commands {1298,929859,497716} : {executing,"cat EUnitFile"} {1298,929859,503066} : "1,2,3" {1298,929859,503118} : no_more_commands {1298,929859,503120} : building_result_tgz {1298,929859,503188} : done [1.105 s] ok wn_job_layer_tests: local_test_ (Canceled in queue)...ok wn_job_layer_tests: local_test_ (Done Job Stored in file layer)...[0.607 s] ok [done in 2.766 s] [done in 2.766 s] ======================================================= All 5 tests passed.
This was fixed by constraining the previous test (and making it more actual to what it should test)
queues_on_resource_types_amount() -> wn_resource_layer:register(#wn_resource{name = "Laptop", type = [{a,0},{b,0}], resides = node() }), Queued = fun() -> wn_resource_layer:queued(node(),"Laptop") end, Job1 = #wn_job{id = "JobId",files = [],resources = [a]}, Job2 = Job1#wn_job{id = "JobId2", resources = [a]}, Job3 = Job1#wn_job{id = "JobId3", resources = [a,b]}, wn_job_layer:register(Job1), ?assertMatch({ok,[{a,[Job1]},{b,[]}]},Queued()), wn_job_layer:register(Job2), ?assertMatch({ok,[{a,[Job1, Job2]},{b,[]}]},Queued()), wn_job_layer:register(Job3), ?assertMatch({ok,[{b,[Job3]}, {a,[Job1,Job2,Job3]} ]},Queued()).
This story has now ended – with the following output of ‘make full’
zen:worker_net-0.1 zenon$ make full erlc -pa . -o ebin/ src/*.erl test/*.erl erl -pa ebin/ -eval 'eunit:test(wn_resource_layer,[verbose]), init:stop().' Erlang R14B (erts-5.8.1) [source] [smp:4:4] [rq:4] [async-threads:0] [hipe] [kernel-poll:false] Eshell V5.8.1 (abort with ^G) 1> ======================== EUnit ======================== module 'wn_resource_layer' module 'wn_resource_layer_tests' wn_resource_layer_tests: local_resource_test_ (Can register resources locally)...[0.001 s] ok wn_resource_layer_tests: register_distributed (Can Register Distributed)...[0.004 s] ok wn_resource_layer_tests: register_restart_register (Can Register, Restart and Register)...[0.010 s] ok wn_resource_layer_tests: register_deregister (Can Register, Deregister and Register)...[0.008 s] ok [done in 0.894 s] [done in 0.894 s] ======================================================= All 4 tests passed. erl -pa ebin/ -eval 'eunit:test(wn_file_layer,[verbose]), init:stop().' Erlang R14B (erts-5.8.1) [source] [smp:4:4] [rq:4] [async-threads:0] [hipe] [kernel-poll:false] Eshell V5.8.1 (abort with ^G) 1> ======================== EUnit ======================== module 'wn_file_layer' module 'wn_file_layer_tests' wn_file_layer_tests: file_layer_local_test_ (Can store file locally)...[0.001 s] ok wn_file_layer_tests: file_layer_local_test_ (Can retrieve files locally)...[0.002 s] ok wn_file_layer_tests: file_layer_local_test_ (Can delete files locally)...[0.001 s] ok wn_file_layer_tests: can_store_distributed (Can store file distributed)...[0.016 s] ok wn_file_layer_tests: can_retrieve_distributed (Can retrieve file distributed)...[0.017 s] ok wn_file_layer_tests: can_delete_distributed (Can delete file distributed)...[0.018 s] ok wn_file_layer_tests: must_retain (Must retain information between node kill and node restart)...[0.321 s] ok [done in 1.667 s] [done in 1.667 s] ======================================================= All 7 tests passed. erl -pa ebin/ -eval 'eunit:test(wn_job_layer,[verbose]), init:stop().' Erlang R14B (erts-5.8.1) [source] [smp:4:4] [rq:4] [async-threads:0] [hipe] [kernel-poll:false] Eshell V5.8.1 (abort with ^G) 1> ======================== EUnit ======================== module 'wn_job_layer' module 'wn_job_layer_tests' wn_job_layer_tests: local_test_ (Can register locally)...[0.004 s] ok wn_job_layer_tests: local_test_ (Executed locally)...{1299,10791,575180} : file_fetching_done {1299,10791,575182} : executing_commands {1299,10791,575370} : {executing,"more EUnitFile"} {1299,10791,582067} : "1,2,3" {1299,10791,582181} : no_more_commands {1299,10791,582183} : building_result_tgz {1299,10791,582278} : done [1.005 s] ok wn_job_layer_tests: local_test_ (Executed queue)...{1299,10792,587205} : file_fetching_done {1299,10792,587208} : executing_commands {1299,10792,587299} : {executing,"file EunitFile"} {1299,10792,594990} : "EunitFile: ASCII text" {1299,10792,595034} : no_more_commands {1299,10792,595037} : building_result_tgz {1299,10792,595117} : done {1299,10792,598436} : file_fetching_done {1299,10792,598446} : executing_commands {1299,10792,598525} : {executing,"cat EUnitFile"} {1299,10792,603488} : "1,2,3" {1299,10792,603533} : no_more_commands {1299,10792,603536} : building_result_tgz {1299,10792,603593} : done [1.104 s] ok wn_job_layer_tests: local_test_ (Queueus on resource type amount)...ok wn_job_layer_tests: local_test_ (Canceled in queue)...ok wn_job_layer_tests: local_test_ (Done Job Stored in file layer)...[0.609 s] ok [done in 2.773 s] [done in 2.773 s] ======================================================= All 6 tests passed. dialyzer src/*.erl test/*.erl Checking whether the PLT /Users/zenon/.dialyzer_plt is up-to-date... yes Proceeding with analysis... Unknown functions: eunit:test/1 done in 0m5.21s done (passed successfully)
The status of the stories is now
Done stories
“I want to be able to describe a job as
- A set of Files [#wn_file{}]
- The possible resource types needed (disjunction)
- A list of commands for running the job
- A timeout the job is given while running (in seconds)
- A job id – primary key for the job“
“I want to be able to register a job in the job layer, through any node.“
“Several jobs should be queued on the same resource type(s), being processed one and one in order as they where queued“
“I want to be able to list all jobs in the system, through any node.”
“The output generated by a job should be possible to see in realtime, through any node and stored to logfiles.”
“A job must be possible to cancel before it starts running, through any node.“
“Once a job is done, the result should be stored in the file layer, together with the logs.“
Not done stories
“I want to be able to delete a job once it’s done, through any node.”
In the next post, I shall write the test and implementation of the current final story on the job layer – to delete a job from any node – once it’s done.
Cheers
/G
Erlang, Testing and TDD 