Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

implement dump=protocol for job plugin

  • Loading branch information...
commit 041e001832d284adefa65465851d75f4ca638dab 1 parent c9f2b9b
Nicolas Niclausse nniclausse authored
8 include/ts_job.hrl
@@ -53,7 +53,13 @@
53 53
54 54 -record(job_session, {
55 55 jobid,
56   - submission_time
  56 + owner,
  57 + submission_time,
  58 + queue_time,
  59 + start_time,
  60 + end_time,
  61 + dump,
  62 + status
57 63 }).
58 64
59 65
14 src/tsung/ts_job.erl
@@ -132,8 +132,14 @@ subst(Job=#job{duration=D,req=Req,jobid=Id}, DynVars) ->
132 132 jobid=ts_search:subst(Id,DynVars)}.
133 133
134 134
135   -dump(A,B) ->
136   - ts_plugin:dump(A,B).
  135 +dump(protocol,{none,#job_session{jobid=JobId,owner=Owner,submission_time=Sub,queue_time=Q,
  136 + start_time=Start,end_time=E,status=Status},Name,_,_})->
  137 + {R,_}=lists:mapfoldl(fun(A,Acc) -> {integer_to_list(round(ts_utils:elapsed(Acc,A))),A} end,Sub,[Q,Start,E]),
  138 + Date=integer_to_list(round(ts_utils:time2sec_hires(Sub))),
  139 + Data=ts_utils:join(";",[JobId,Name,Date]++R++[Status]),
  140 + ts_mon:dump({protocol, Owner, Data });
  141 +dump(_P,_Args) ->
  142 + ok.
137 143
138 144 %% @spec parse(Data::client_data(), State) -> {NewState, Opts, Close}
139 145 %% State = #state_rcv{}
@@ -146,7 +152,7 @@ dump(A,B) ->
146 152 %% Setting Close to true will cause tsung to close the connection to
147 153 %% the server.
148 154 %% @end
149   -parse({os, cmd, _Args, Res},State=#state_rcv{session=S}) when is_list(Res)->
  155 +parse({os, cmd, _Args, Res},State=#state_rcv{session=S,dump=Dump}) when is_list(Res)->
150 156 ?LOGF("os:cmd result: ~p",[Res],?DEB),
151 157 %% oarsub output:
152 158 %% [ADMISSION RULE] Modify resource description with type constraints
@@ -156,7 +162,7 @@ parse({os, cmd, _Args, Res},State=#state_rcv{session=S}) when is_list(Res)->
156 162 case lists:last(Lines) of
157 163 "OAR_JOB_ID="++ID ->
158 164 ?LOGF("OK,job id is ~p",[ID],?INFO),
159   - ts_job_notify:monitor({ID,self(),S#job_session.submission_time, now()}),
  165 + ts_job_notify:monitor({ID,self(),S#job_session.submission_time, now(),Dump}),
160 166 {State#state_rcv{ack_done=true,datasize=length(Res)}, [], false};
161 167 _ ->
162 168 {State#state_rcv{ack_done=true,datasize=length(Res)}, [], false}
34 src/tsung_controller/ts_job_notify.erl
@@ -30,6 +30,7 @@
30 30 -behaviour(gen_server).
31 31
32 32 -include("ts_profile.hrl").
  33 +-include("ts_job.hrl").
33 34
34 35 %% API
35 36 -export([start_link/0]).
@@ -47,7 +48,6 @@
47 48 acceptloop_pid, % The PID of the companion process that blocks
48 49 jobs}).
49 50
50   --record(job, {id,owner,queue_time,start_time}).
51 51
52 52 %%%===================================================================
53 53 %%% API
@@ -66,8 +66,8 @@ start_link() ->
66 66 listen(Port) ->
67 67 gen_server:cast({global, ?MODULE}, {listen, Port}).
68 68
69   -monitor({JobID, OwnerPid, StartTime, QueuedTime}) ->
70   - gen_server:cast({global, ?MODULE}, {monitor, {JobID, OwnerPid, StartTime, QueuedTime}}).
  69 +monitor({JobID, OwnerPid, StartTime, QueuedTime, Dump}) ->
  70 + gen_server:cast({global, ?MODULE}, {monitor, {JobID, OwnerPid, StartTime, QueuedTime,Dump}}).
71 71
72 72 demonitor({JobID}) ->
73 73 gen_server:cast({global, ?MODULE}, {monitor, {JobID}}).
@@ -94,7 +94,7 @@ init([]) ->
94 94 ?LOG("Starting~n",?DEB),
95 95 case global:whereis_name(ts_config_server) of
96 96 undefined ->
97   - {ok, #state{jobs=ets:new(jobs,[{keypos, #job.id}])}};
  97 + {ok, #state{jobs=ets:new(jobs,[{keypos, #job_session.jobid}])}};
98 98 _Pid ->
99 99 ?LOG("Config server is alive !~n",?DEB),
100 100 case ts_config_server:get_jobs_state() of
@@ -104,7 +104,7 @@ init([]) ->
104 104 {ok, NewState};
105 105 Else ->
106 106 ?LOGF("Got this from config server:~p~n",[Else],?DEB),
107   - {ok, #state{jobs=ets:new(jobs,[{keypos, #job.id}])}}
  107 + {ok, #state{jobs=ets:new(jobs,[{keypos, #job_session.jobid}])}}
108 108 end
109 109 end.
110 110
@@ -145,9 +145,9 @@ handle_call(_Request, _From, State) ->
145 145 %% {stop, Reason, State}
146 146 %% @end
147 147 %%--------------------------------------------------------------------
148   -handle_cast({monitor, {JobID, OwnerPid, SubmitTS, QueuedTS}}, State=#state{jobs=Jobs}) ->
  148 +handle_cast({monitor, {JobID, OwnerPid, SubmitTS, QueuedTS,Dump}}, State=#state{jobs=Jobs}) ->
149 149 ?LOGF("monitoring job ~p from pid ~p~n",[JobID,OwnerPid],?DEB),
150   - ets:insert(Jobs,#job{id=JobID,owner=OwnerPid, queue_time=QueuedTS}),
  150 + ets:insert(Jobs,#job_session{jobid=JobID,owner=OwnerPid, submission_time=SubmitTS, queue_time=QueuedTS,dump=Dump}),
151 151 SubmitTime=ts_utils:elapsed(SubmitTS,QueuedTS),
152 152 ts_mon:add([{sum,job_queued,1},{sample,tr_job_submit,SubmitTime}]),
153 153 {noreply, State};
@@ -210,31 +210,33 @@ handle_info({tcp, Socket, Data}, State=#state{jobs=Jobs}) ->
210 210 ?LOGF("Job owner of ~p is unknown",[Id],?NOTICE);
211 211 [Job] ->
212 212 Now=now(),
213   - Queued=ts_utils:elapsed(Job#job.queue_time,Now),
  213 + Queued=ts_utils:elapsed(Job#job_session.queue_time,Now),
214 214 ts_mon:add([{sample,tr_job_wait,Queued},{sum,job_running,1}, {sum,job_queued,-1}]),
215   - ets:update_element(Jobs,Id,{#job.start_time,Now})
  215 + ets:update_element(Jobs,Id,{#job_session.start_time,Now})
216 216 end;
217   - [Id, _Name, "END"|_] ->
  217 + [Id, Name, "END"|_] ->
218 218 case ets:lookup(Jobs,Id) of
219 219 [] ->
220 220 ?LOGF("Job owner of ~p is unknown",[Id],?NOTICE);
221 221 [Job]->
222 222 Now=now(),
223   - Duration=ts_utils:elapsed(Job#job.start_time,Now),
  223 + Duration=ts_utils:elapsed(Job#job_session.start_time,Now),
224 224 ts_mon:add([{sample,tr_job_duration,Duration},{sum,job_running,-1}, {sum,ok_job ,1}]),
  225 + ts_job:dump(Job#job_session.dump,{none,Job#job_session{end_time=Now,status="ok"},Name,undefined,undefined}),
225 226 ets:delete_object(Jobs,Job),
226   - check_jobs(Jobs,Job#job.owner)
  227 + check_jobs(Jobs,Job#job_session.owner)
227 228 end;
228   - [Id, _Name, "ERROR"|_] ->
  229 + [Id, Name, "ERROR"|_] ->
229 230 case ets:lookup(Jobs,Id) of
230 231 [] ->
231 232 ?LOGF("Job owner of ~p is unknown",[Id],?NOTICE);
232 233 [Job]->
233 234 Now=now(),
234   - Duration=ts_utils:elapsed(Job#job.start_time,Now),
  235 + Duration=ts_utils:elapsed(Job#job_session.start_time,Now),
235 236 ts_mon:add([{sample,tr_job_duration,Duration},{sum,job_running,-1}, {sum,error_job,1}]),
  237 + ts_job:dump(Job#job_session.dump,{none,Job#job_session{end_time=Now,status="error"},Name,undefined,undefined}),
236 238 ets:delete_object(Jobs,Job),
237   - check_jobs(Jobs,Job#job.owner)
  239 + check_jobs(Jobs,Job#job_session.owner)
238 240 end;
239 241 [_Id, _Name, "INFO"|_] ->
240 242 ok;
@@ -294,7 +296,7 @@ code_change(_OldVsn, State, _Extra) ->
294 296 %%%===================================================================
295 297
296 298 check_jobs(Jobs,Pid)->
297   - case ets:match_object(Jobs, #job{owner=Pid, _='_'}) of
  299 + case ets:match_object(Jobs, #job_session{owner=Pid, _='_'}) of
298 300 [] ->
299 301 ?LOGF("no jobs for pid ~p~n",[Pid],?DEB),
300 302 Pid ! {erlang, ok, nojobs};

0 comments on commit 041e001

Please sign in to comment.
Something went wrong with that request. Please try again.