聶永的博客

前言

接著上文，tsung一旦啟動，主從節點之間需要協調分配資源，完成分布式壓測任務。

如何啟動Tsung壓測從機

Erlang SDK提供了從機啟動方式：

slave:start(Host, Node, Opts)

啟動從機需要借助于免登陸形式遠程終端，比如SSH（后續會討論SSH存在不足，以及全新的替代品），需要自行配置。

<client host="client_100" maxusers="60000" weight="1">
    <ip value="10.10.10.100"/>
</client>

• host屬性對應value為從機主機名稱：client_100
• Node節點名稱由tsung_controller組裝，類似于 tsung10@client_100
• Opts表示相關參數
• 一個物理機器，可以存在多個tsung從機實例
• 一個tsung從機實例對應一個tsung client

簡單翻譯一下：slave:start(client_100, 'tsung10@client_100', Opts)

從機需要關閉時，就很簡單了：

slave:stop(Node)

當然若主機中途掛掉，從機也會自動自殺掉自身。

啟動tsung client方式

Tsung主機啟動從機成功，從機和主機就可以Erlang節點進程之間進行方法調用和消息傳遞。潛在要求是，tsung編譯后beam文件能夠在Erlang運行時環境中能夠訪問到，這個和Java Classpath一致原理。

rpc:multicall(RemoteNodes,tsung,start,[],?RPC_TIMEOUT)

到此為止，一個tsung client實例成功運行。

• tsung client實例生命周期結束，不會導致從機實例主動關閉
• tsung slave提供了運行時環境，tsung client是業務
• tsung slave和tsung client關系是1 : 1關系，很多時候為了理解方便，不會進行嚴格區分

壓測目標

明白了主從啟動方式，下面討論壓測目標，比如50萬用戶的量，根據給出的壓測從機列表，進行任務分配。

壓測目標配置

tsung壓測xml配置文件，load元素可以配置總體任務生成的信息。

<load>
    <arrivalphase phase="1" duration="60" unit="minute">
        <!--users maxnumber="500000" interarrival="0.004" unit="second"></users-->
        <users maxnumber="500000" arrivalrate="250" unit="second"></users>
    </arrivalphase>
</load>

• 定義一個最終壓力產生可以持續60分鐘壓測場景， 上限用戶量為50萬
• arrivalphase duration屬性持續時長表示生成壓測用戶可消費總體時間60分鐘，即為T1
• users元素其屬性表示單位時間內（這里單位時間為秒）產生用戶數為250個
• 50萬用戶，將在2000秒(約34分鐘)內生成，耗時時長即為T2
• T2小于arrivalphase定義的用戶生成階段持續時間T1
• 若T2時間后（34分鐘)后因為產生用戶數已經達到了上限，將不再產生新的用戶，知道整個壓測結束
• 若 T1 小于 T2，則50萬用戶很難達到，因此T1時間要設置長一些

從節點信息配置

所說從節點也是壓測客戶端，需要配置clients元素：

<clients>
    <client host="client_100" maxusers="60000" weight="1">
        <ip value="10.10.10.100"/>
    </client>

    ......

    <client host="client_109" maxusers="120000" weight="2">
        <ip value="10.10.10.109"></ip>
        <ip value="10.10.10.119"></ip>
    </client>
</clients>

1. 單個client支持多個IP，用于突破單個IP對外建立連接數的限制(后續會講到）
2. xml所定義的一個cliet元素，可能被分裂出若干從機實例(即tsung client)，1 : N

根據CPU數量分裂tsung client實例情況

在《Tsung Documentation》給出了建議，一個CPU一個tsung client實例：

Note: Even if an Erlang VM is now able to handle several CPUs (erlang SMP), benchmarks shows that it’s more efficient to use one VM per CPU (with SMP disabled) for tsung clients. Only the controller node is using SMP erlang.
Therefore, cpu should be equal to the number of cores of your nodes. If you prefer to use erlang SMP, add the -s option when starting tsung (and don’t set cpu in the config file).

• 默認策略, 一個tsung client對應一個CPU，若不設置CPU屬性，默認值就是1
• 一個cpu對應一個tsung client，N個CPU，N個tsung client
• 共同分擔權重，每一個分裂的tsung client權重 Weight/N
• 一旦設置cpu屬性，無論Tsung啟動時是否攜帶-s參數設置共享CPU，都會
  • 自動分裂CPU個tsung client實例
  • 每一個實例權重為Weight/CPU

%% add a new client for each CPU
lists:duplicate(CPU,#client{host     = Host,
                            weight   = Weight/CPU,
                            maxusers = MaxUsers})

若要設置單個tsung client實例共享多個CPU（此時不要設置cpu屬性啦），需要在tsung啟動時添加-s參數，tsung client被啟動時，smp屬性被設置成auto：

-smp auto +A 8

這樣從機就只有一個tsung client實例了，不會讓人產生困擾。若是臨時租借從機，建議啟動時使用-s參數，并且要去除cpu屬性設置，這樣才能夠自動共享所有CPU核心。

從機分配用戶過多，一樣會分裂新的tsung client實例

假設client元素配置maxusers數量為1K，那么實際上被分配數量為10K(壓測人數多，壓測從機少)時，那么tsung_controller會繼續分裂新的tsung client實例，直到10K用戶數量完成。

<client host="client_98" maxusers="1000" weight="1">
    <ip value="10.10.10.98"></ip>
</client>

tsung client分配的數量超過自身可服務上限用戶時（這里設置的是1K）時，關閉自身。

launcher(_Event, State=#launcher{nusers = 0, phases = [] }) ->
    ?LOG("no more clients to start, stop  ~n",?INFO),
    {stop, normal, State};

launcher(timeout, State=#launcher{nusers        = Users,
                                  phase_nusers  = PhaseUsers,
                                  phases        = Phases,
                                  phase_id      = Id,
                                  started_users = Started,
                                  intensity     = Intensity}) ->
    BeforeLaunch = ?NOW,
    case do_launch({Intensity,State#launcher.myhostname,Id}) of
        {ok, Wait} ->
            case check_max_raised(State) of
                true ->
                    %% let the other beam starts and warns ts_mon
                    timer:sleep(?DIE_DELAY),
                    {stop, normal, State};
                false->
                    ......
            end;
        error ->
            % retry with the next user, wait randomly a few msec
            RndWait = random:uniform(?NEXT_AFTER_FAILED_TIMEOUT),
            {next_state,launcher,State#launcher{nusers = Users-1} , RndWait}
    end.

tsung_controller接收從節點退出通知，但分配總數沒有完成，會啟動新的tsung client實例（一樣先啟動從節點，然后再啟動tsung client實例）。整個過程串行方式循環，直到10K用戶數量完成：

%% start a launcher on a new beam with slave module
handle_cast({newbeam, Host, Arrivals}, State=#state{last_beam_id = NodeId, config=Config, logdir = LogDir}) ->
    Args = set_remote_args(LogDir,Config#config.ports_range),
    Seed = Config#config.seed,
    Node = remote_launcher(Host, NodeId, Args),
    case rpc:call(Node,tsung,start,[],?RPC_TIMEOUT) of
        {badrpc, Reason} ->
            ?LOGF("Fail to start tsung on beam ~p, reason: ~p",[Node,Reason], ?ERR),
            slave:stop(Node),
            {noreply, State};
        _ ->
            ts_launcher_static:stop(Node), % no need for static launcher in this case (already have one)
            ts_launcher:launch({Node, Arrivals, Seed}),
            {noreply, State#state{last_beam_id = NodeId+1}}
    end;

tsung client分配用戶數

一個tsung client分配的用戶數，可以理解為會話任務數。Tsung以終端可以模擬的用戶為維度進行定義壓測。

所有配置tsung client元素（設置M1）權重相加之和為總權重TotalWeight，用戶總數為MaxMember，一個tsung client實例（總數設為M2）分配的模擬用戶數可能為：

MaxMember*(Weight/TotalWeight)

需要注意：
- M2 >= M1
- 若壓測階段<arrivalphase元素配置duration值過小，小于最終用戶50萬用戶按照每秒250速率耗時時間，最終分配用戶數將小于期望值

只有一臺物理機的tsung master啟動方式

<clients>
  <client host="localhost" use_controller_vm="true"/>
</clients>

沒有物理從機，主從節點都在一臺機器上，需要設置use_controller_vm="true"。相比tsung集群，單一節點tsung啟動就很簡單，主從之間不需要SSH通信，直接內部調用。

local_launcher([Host],LogDir,Config) ->
    ?LOGF("Start a launcher on the controller beam ~p~n", [Host], ?NOTICE),
    LogDirEnc = encode_filename(LogDir),
    %% set the application spec (read the app file and update some env. var.)
    {ok, {_,_,AppSpec}} = load_app(tsung),
    {value, {env, OldEnv}} = lists:keysearch(env, 1, AppSpec),
    NewEnv = [ {debug_level,?config(debug_level)}, {log_file,LogDirEnc}],
    RepKeyFun = fun(Tuple, List) ->  lists:keyreplace(element(1, Tuple), 1, List, Tuple) end,
    Env = lists:foldl(RepKeyFun, OldEnv, NewEnv),
    NewAppSpec = lists:keyreplace(env, 1, AppSpec, {env, Env}),

    ok = application:load({application, tsung, NewAppSpec}),
    case application:start(tsung) of
        ok ->
            ?LOG("Application started, activate launcher, ~n", ?INFO),
            application:set_env(tsung, debug_level, Config#config.loglevel),
            case Config#config.ports_range of
                {Min, Max} ->
                    application:set_env(tsung, cport_min, Min),
                    application:set_env(tsung, cport_max, Max);
                undefined ->
                    ""
            end,
            ts_launcher_static:launch({node(), Host, []}),
            ts_launcher:launch({node(), Host, [], Config#config.seed}),
            1 ;
        {error, Reason} ->
            ?LOGF("Can't start launcher application (reason: ~p) ! Aborting!~n",[Reason],?EMERG),
            {error, Reason}
    end.

用戶生成控制

用戶和會話控制

每一個tsung client運行著一個ts_launch/ts_launch_static本地注冊模塊，掌控終端模擬用戶生成和會話控制。

• 向主節點ts_config_server請求隸屬于當前從機節點的會話信息
• 啟動模擬終端用戶ts_client
• 控制下一個模擬終端用戶ts_client需要等待時間，也是控制從機用戶生成速度
• 執行是否需要切換到新的階段會話
• 控制模擬終端用戶是否已經達到了設置的maxusers上限
  • 到上限，自身使命完成，關閉自身
• 源碼位于 tsung-1.6.0/src/tsung 目錄下

主機按照xml配置生成全局用戶產生速率，從機按照自身權重分配的速率進行單獨控制，這也是任務分解的具體呈現。

用戶生成速度控制

在Tsung中用戶生成速度稱之為強度，根據所配置的load屬性進行配置

<load>
    <arrivalphase phase="1" duration="60" unit="minute">
        <users maxnumber="500000" arrivalrate="250" unit="second"></users>
    </arrivalphase>
</load>

關鍵屬性：

• interarrival，生成壓測用戶的時間間隔
• arrivalrate：單位時間內生成用戶數量
• 兩者最終都會被轉換為生成用戶強度系數值是0.25
• 這個是總的強度值，但需要被各個tsung client分解

parse(Element = #xmlElement{name=users, attributes=Attrs},
      Conf = #config{arrivalphases=[CurA | AList]}) ->

    Max = getAttr(integer,Attrs, maxnumber, infinity),
    ?LOGF("Maximum number of users ~p~n",[Max],?INFO),

    Unit  = getAttr(string,Attrs, unit, "second"),
    Intensity = case {getAttr(float_or_integer,Attrs, interarrival),
                      getAttr(float_or_integer,Attrs, arrivalrate)  } of
                    {[],[]} ->
                        exit({invalid_xml,"arrival or interarrival must be specified"});
                    {[], Rate}  when Rate > 0 ->
                        Rate / to_milliseconds(Unit,1);
                    {InterArrival,[]} when InterArrival > 0 ->
                        1/to_milliseconds(Unit,InterArrival);
                    {_Value, _Value2} ->
                        exit({invalid_xml,"arrivalrate and interarrival can't be defined simultaneously"})
                end,
    lists:foldl(fun parse/2,
        Conf#config{arrivalphases = [CurA#arrivalphase{maxnumber = Max,
                                                        intensity=Intensity}
                               |AList]},
                Element#xmlElement.content);

tsung_controller對每一個tsung client生成用戶強度分解為 ClientIntensity = PhaseIntensity * Weight / TotalWeight，而1000 * ClientIntensity就是易讀的每秒生成用戶速率值。

get_client_cfg(Arrival=#arrivalphase{duration = Duration,
                                     intensity= PhaseIntensity,
                                     curnumber= CurNumber,
                                     maxnumber= MaxNumber },
               {TotalWeight,Client,IsLast} ) ->
    Weight = Client#client.weight,
    ClientIntensity = PhaseIntensity * Weight / TotalWeight,
    NUsers = round(case MaxNumber of
                       infinity -> %% only use the duration to set the number of users
                           Duration * ClientIntensity;
                       _ ->
                           TmpMax = case {IsLast,CurNumber == MaxNumber} of
                                        {true,_} ->
                                            MaxNumber-CurNumber;
                                        {false,true} ->
                                            0;
                                        {false,false} ->
                                            lists:max([1,trunc(MaxNumber * Weight / TotalWeight)])
                                    end,
                           lists:min([TmpMax, Duration*ClientIntensity])
                   end),
    ?LOGF("New arrival phase ~p for client ~p (last ? ~p): will start ~p users~n",
          [Arrival#arrivalphase.phase,Client#client.host, IsLast,NUsers],?NOTICE),
    {Arrival#arrivalphase{curnumber=CurNumber+NUsers}, {ClientIntensity, NUsers, Duration}}.

前面講到每一個tsung client被分配用戶數公式為：min(Duration * ClientIntensity, MaxNumber * Weight / TotalWeight)：

• 避免總人數超出限制
• 階段Phase持續時長所產生用戶數和tsung client分配用戶數不至于產生沖突，一種協調策略

再看一下launch加載一個終端用戶時，會自動根據當前分配用戶生成壓力系數獲得ts_stats:exponential(Intensity)下一個模擬用戶產生等待生成的最長時間，單位為毫秒。

do_launch({Intensity, MyHostName, PhaseId})->
    %%Get one client
    %%set the profile of the client
    case catch ts_config_server:get_next_session({MyHostName, PhaseId} ) of
        {'EXIT', {timeout, _ }} ->
            ?LOG("get_next_session failed (timeout), skip this session !~n", ?ERR),
            ts_mon:add({ count, error_next_session }),
            error;
        {ok, Session} ->
            ts_client_sup:start_child(Session),
            X = ts_stats:exponential(Intensity),
            ?DebugF("client launched, wait ~p ms before launching next client~n",[X]),
            {ok, X};
        Error ->
            ?LOGF("get_next_session failed for unexpected reason [~p], abort !~n", [Error],?ERR),
            ts_mon:add({ count, error_next_session }),
            exit(shutdown)
    end.

ts_stats:exponential邏輯引入了指數計算：

exponential(Param) ->
    -math:log(random:uniform())/Param.

繼續往下看吧，隱藏了部分無關代碼：

launcher(timeout, State=#launcher{nusers        = Users,
                                  phase_nusers  = PhaseUsers,
                                  phases        = Phases,
                                  phase_id      = Id,
                                  started_users = Started,
                                  intensity     = Intensity}) ->
    BeforeLaunch = ?NOW,
    case do_launch({Intensity,State#launcher.myhostname,Id}) of
        {ok, Wait} ->
                            ...
                        {continue} ->
                            Now=?NOW,
                            LaunchDuration = ts_utils:elapsed(BeforeLaunch, Now),
                            %% to keep the rate of new users as expected,
                            %% remove the time to launch a client to the next
                            %% wait.
                            NewWait = case Wait > LaunchDuration of
                                          true -> trunc(Wait - LaunchDuration);
                                          false -> 0
                                      end,
                            ?DebugF("Real Wait = ~p (was ~p)~n", [NewWait,Wait]),
                            {next_state,launcher,State#launcher{nusers = Users-1, started_users=Started+1} , NewWait}
                            ...
        error ->
            % retry with the next user, wait randomly a few msec
            RndWait = random:uniform(?NEXT_AFTER_FAILED_TIMEOUT),
            {next_state,launcher,State#launcher{nusers = Users-1} , RndWait}
    end.

下一個用戶生成需要等待Wait - LaunchDuration毫秒時間。

給出一個采樣數據，只有一個從機，并且用戶產生速度1秒一個，共產生10個用戶：

<load>
    <arrivalphase phase="1" duration="50" unit="minute">
        <users maxnumber="10" interarrival="1" unit="second"/>
    </arrivalphase>
</load>

采集日志部分，記錄了Wait時間值，其實總體時間還需要加上LaunchDuration（雖然這個值很小）：

ts_launcher:(7:<0.63.0>) client launched, wait 678.5670934164623 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 810.2982455546687 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 1469.2208436232288 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 986.7202548184069 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 180.7484423006169 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 1018.9190235965457 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 1685.0156394273606 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 408.53992361334065 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 204.40900996137086 ms before launching next client
ts_launcher:(7:<0.63.0>) client launched, wait 804.6040921461512 ms before launching next client

總體來說，每一個用戶生成間隔間不是固定值，是一個大約值，有偏差，但接近于目標設定（1000毫秒生成一個用戶標準間隔）。

執行模擬終端用戶會話流程

關于會話的說明：

• 一個session元素中的定義一系列請求-響應等交互行為稱之為一次完整會話
• 一個模擬用戶需要執行一次完整會話，然后生命周期完成，然后結束

模擬終端用戶模塊是ts_client（狀態機），掛載在ts_client_sup下，由ts_launcher/ts_launcher_static調用ts_client_sup:start_child(Session)啟動，是壓測任務的最終執行者，承包了所有臟累差的活：

• 所有下一步需要執行的會話指令都需要向主機的ts_config_server請求
• 執行會話指令
• 具體協議調用相應協議插件，比如ts_mqtt組裝會話消息
• 建立網絡Socket連接，封裝眾多網絡通道
• 發送請求數據，處理響應
• 記錄并發送監控數據和日志

?

小結

簡單梳理主從之間啟動方式，從機數量分配策略，以具體壓測任務如何在從機上分配和運行等內容。