anchor.diff:      end,
       TimeFinish = os:timestamp(),
       mzb_metrics:notify({"latency", histogram}, timer:now_diff(TimeFinish, TimeStart)),
  -    {nil, State}.
  +    {nil, NewState}.
   
   run(Command, _Opts, WorkerId) ->
       {ok, Pid, _OsPid} = exec:run(Command, [monitor, stdout, stderr,
  
positive.code: -export([
      initial_state/0,
      metrics/0,
      execute/3
  ]).
  
  initial_state() -> "".
  
  metrics() ->
      [
  
negative.code: 
  get_data(Port, WorkerId, _Buffer) ->
      receive
          {stdout, _Pid, Bytes} ->
              lager:info("Worker ~p output: ~s", [WorkerId, Bytes]),
              get_data(Port, WorkerId, []);
          {stderr, _Pid, Bytes} ->
              lager:error("Worker ~p error: ~s", [WorkerId, Bytes]),
              get_data(Port, WorkerId, []);
          {'DOWN', _ , _, _, {exit_status, Code}} -> Code;
          {'DOWN', _ , _, _, normal} -> 0
      end.
  
  statsd(Opts) ->
      Port =
          case proplists:get_value(port, Opts) of
              undefined -> ?DEFAULT_STATSD_PORT;
              P -> P
          end,
      {ok, ListenSocket} = gen_udp:open(Port, [binary, {active, false}]),
      accept(ListenSocket).
  
  accept(Socket) ->
      inet:setopts(Socket, [{active, once}]),
      receive
          {udp, Socket, _Host, _Port, Binary} ->
              parse(Binary),
              accept(Socket)
      end.
  
  parse(<<>>) -> ok;
  parse(Bin) ->
      lists:map(fun report/1, binary:split(Bin, <<"\n">>, [global])).
  
  report(<<>>) -> ok;
  report(Bin) ->
      case binary:split(Bin, [<<"|">>, <<":">>], [global]) of
          [Metric, Value, Type|_] ->
              MetricStr = binary_to_list(Metric),
              {TypeAtom, ValueInt} =
                  case Type of
                      <<"c">> -> {counter, parse_num(Value)};
                      <<"g">> -> {gauge, parse_num(Value)};
                      _ -> lager:info("Unknown type ~p", [Type]), gauge
                  end,
              mzb_metrics:notify({MetricStr, TypeAtom}, ValueInt);
          _ -> lager:info("Unknown format: ~s", [Bin])
      end.
  
  parse_num(Bin) ->
      try erlang:binary_to_integer(Bin) of
          V -> V
      catch
          _:_ ->
              erlang:round(erlang:binary_to_float(Bin))
      end.
  
  start_statsd(State = #{statsd_opts := Opts}) ->
      erlang:spawn(?MODULE, statsd, [Opts]),
      State.
  
  