zf

zenflows testing
git clone https://s.sonu.ch/~srfsh/zf.git
Log | Files | Refs | Submodules | README | LICENSE

cowboy_telemetry_h.erl (2948B)

      1 -module(cowboy_telemetry_h).
      2 -behavior(cowboy_stream).
      3 
      4 -export([init/3]).
      5 -export([data/4]).
      6 -export([info/3]).
      7 -export([terminate/3]).
      8 -export([early_error/5]).
      9 
     10 init(StreamID, Req, Opts) ->
     11     telemetry:execute(
     12         [cowboy, request, start],
     13         #{system_time => erlang:system_time()},
     14         #{streamid => StreamID, req => Req}),
     15     cowboy_metrics_h:init(StreamID, Req, add_metrics_callback(Opts)).
     16 
     17 info(StreamID, Info, State) ->
     18     cowboy_metrics_h:info(StreamID, Info, State).
     19 
     20 data(StreamID, IsFin, Data, State) ->
     21     cowboy_metrics_h:data(StreamID, IsFin, Data, State).
     22 
     23 terminate(StreamID, Reason, State) ->
     24     cowboy_metrics_h:terminate(StreamID, Reason, State).
     25 
     26 early_error(StreamID, Reason, PartialReq, Resp, Opts) ->
     27     cowboy_metrics_h:early_error(StreamID, Reason, PartialReq, Resp, add_metrics_callback(Opts)).
     28 
     29 %
     30 
     31 add_metrics_callback(Opts) ->
     32     maps:put(metrics_callback, fun metrics_callback/1, Opts).
     33 
     34 metrics_callback(#{early_error_time := Time} = Metrics) when is_number(Time) ->
     35     {RespBodyLength, Metadata} = maps:take(resp_body_length, Metrics),
     36     telemetry:execute(
     37         [cowboy, request, early_error],
     38         #{system_time => erlang:system_time(), resp_body_length => RespBodyLength},
     39         Metadata);
     40 metrics_callback(#{reason := {internal_error, {'EXIT', _, {Reason, Stacktrace}}, _}} = Metrics) ->
     41     telemetry:execute(
     42         [cowboy, request, exception],
     43         measurements(Metrics),
     44         (metadata(Metrics))#{kind => exit, reason => Reason, stacktrace => Stacktrace});
     45 metrics_callback(#{reason := {ErrorType, _, _} = Reason} = Metrics)
     46     when ErrorType == socket_error;
     47          ErrorType == connection_error ->
     48     telemetry:execute(
     49         [cowboy, request, stop],
     50         measurements(Metrics),
     51         (metadata(Metrics))#{error => Reason});
     52 metrics_callback(Metrics) ->
     53     telemetry:execute(
     54         [cowboy, request, stop],
     55         measurements(Metrics),
     56         metadata(Metrics)).
     57 
     58 measurements(Metrics) ->
     59     #{req_body_length := ReqBodyLength, resp_body_length := RespBodyLength} = Metrics,
     60 
     61     #{
     62         duration => duration(req_start, req_end, Metrics),
     63         req_body_duration => duration(req_body_start, req_body_end, Metrics),
     64         resp_duration => duration(resp_start, resp_end, Metrics),
     65         req_body_length => ReqBodyLength,
     66         resp_body_length => RespBodyLength
     67     }.
     68 
     69 metadata(Metrics) ->
     70     #{
     71         pid := Pid,
     72         streamid := Streamid,
     73         req := Req,
     74         resp_headers := RespHeaders,
     75         resp_status := RespStatus,
     76         ref := Ref
     77     } = Metrics,
     78 
     79     #{
     80         pid => Pid,
     81         streamid => Streamid,
     82         req => Req,
     83         resp_headers => RespHeaders,
     84         resp_status => RespStatus,
     85         ref => Ref
     86     }.
     87 
     88 duration(StartKey, EndKey, Metrics) ->
     89     case Metrics of
     90         #{StartKey := Start, EndKey := End} when is_integer(Start), is_integer(End) -> End - Start;
     91         #{} -> 0
     92     end.