cowboy_metrics_h.erl

%% Copyright (c) 2017, Loïc Hoguin <[email protected]>
%%
%% Permission to use, copy, modify, and/or distribute this software for any
%% purpose with or without fee is hereby granted, provided that the above
%% copyright notice and this permission notice appear in all copies.
%%
%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

-module(cowboy_metrics_h).
-behavior(cowboy_stream).

-export([init/3]).
-export([data/4]).
-export([info/3]).
-export([terminate/3]).
-export([early_error/5]).

-type proc_metrics() :: #{pid() => #{
	spawn := integer(),
	exit => integer(),
	reason => any()
}}.

-type metrics() :: #{
	%% The identifier for this listener.
	ref := ranch:ref(),

	%% The pid for this connection.
	pid := pid(),

	%% The streamid also indicates the total number of requests on
	%% this connection (StreamID div 2 + 1).
	streamid := cowboy_stream:streamid(),

	%% @todo Req (filtered)
	%% @todo Resp (filtered?)
	%% @todo Handler...

	%% The terminate reason is always useful.
	reason := cowboy_stream:reason(),

	%% Start/end of the processing of the request.
	%%
	%% This represents the time from this stream handler's init
	%% to terminate. Note that this doesn't indicate the response
	%% has been sent fully, it still may be queued up in a buffer.
	req_start => integer(),
	req_end => integer(),

	%% Start/end of the receiving of the request body.
	%% Begins when the first packet has been received.
	req_body_start => integer(),
	req_body_end => integer(),

	%% Start/end of the sending of the response.
	%% Begins when we send the headers and ends on the final
	%% packet of the response body. If everything is sent at
	%% once these values are identical.
	resp_start => integer(),
	resp_end => integer(),

	%% For early errors all we get is the time we received it.
	early_error_time => integer(),

	%% Start/end of spawned processes. This is where most of
	%% the user code lies, excluding stream handlers. On a
	%% default Cowboy configuration there should be only one
	%% process: the request process.
	procs => proc_metrics(),

	%% Length of the request and response bodies. This does
	%% not include the framing.
	req_body_length => non_neg_integer(),
	resp_body_length => non_neg_integer()
}.
-export_type([metrics/0]).

-record(state, {
	next :: any(),
	callback :: fun((metrics()) -> any()),
	req_filter :: fun((cowboy_req:req()) -> map()),
	req = #{} :: map(),
	ref = undefined :: ranch:ref(),
	req_start :: integer(),
	req_end :: integer(),
	req_body_start :: integer(),
	req_body_end :: integer(),
	resp_start :: integer(),
	resp_end :: integer(),
	procs = #{} :: proc_metrics(),
	req_body_length = 0 :: non_neg_integer(),
	resp_body_length = 0 :: non_neg_integer()
}).

-spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
	-> {[{spawn, pid(), timeout()}], #state{}}.
init(StreamID, Req=#{ref := Ref}, Opts=#{metrics_callback := Fun}) ->
	ReqStart = erlang:system_time(nanosecond),
	{Commands, Next} = cowboy_stream:init(StreamID, Req, Opts),
	{Commands, fold(Commands, #state{next=Next, callback=Fun, ref=Ref, req_start=ReqStart})}.

-spec data(cowboy_stream:streamid(), cowboy_stream:fin(), cowboy_req:resp_body(), State)
	-> {cowboy_stream:commands(), State} when State::#state{}.
data(StreamID, IsFin=fin, Data, State=#state{req_body_start=undefined}) ->
	ReqBody = erlang:system_time(nanosecond),
	do_data(StreamID, IsFin, Data, State#state{
		req_body_start=ReqBody,
		req_body_end=ReqBody,
		req_body_length=byte_size(Data)
	});
data(StreamID, IsFin=fin, Data, State=#state{req_body_length=ReqBodyLen}) ->
	ReqBodyEnd = erlang:system_time(nanosecond),
	do_data(StreamID, IsFin, Data, State#state{
		req_body_end=ReqBodyEnd,
		req_body_length=ReqBodyLen + byte_size(Data)
	});
data(StreamID, IsFin, Data, State=#state{req_body_start=undefined}) ->
	ReqBodyStart = erlang:system_time(nanosecond),
	do_data(StreamID, IsFin, Data, State#state{
		req_body_start=ReqBodyStart,
		req_body_length=byte_size(Data)
	});
data(StreamID, IsFin, Data, State) ->
	do_data(StreamID, IsFin, Data, State).

do_data(StreamID, IsFin, Data, State0=#state{next=Next0}) ->
	{Commands, Next} = cowboy_stream:data(StreamID, IsFin, Data, Next0),
	{Commands, fold(Commands, State0#state{next=Next})}.

-spec info(cowboy_stream:streamid(), any(), State)
	-> {cowboy_stream:commands(), State} when State::#state{}.
info(StreamID, Info={'EXIT', Pid, Reason}, State0=#state{procs=Procs}) ->
	ProcEnd = erlang:system_time(nanosecond),
	P = maps:get(Procs, Pid),
	State = State0#state{procs=Procs#{Pid => P#{
		exit => ProcEnd,
		reason => Reason
	}}},
	do_info(StreamID, Info, State);
info(StreamID, Info, State) ->
	do_info(StreamID, Info, State).

do_info(StreamID, Info, State0=#state{next=Next0}) ->
	{Commands, Next} = cowboy_stream:info(StreamID, Info, Next0),
	{Commands, fold(Commands, State0#state{next=Next})}.

fold([], State) ->
	State;
fold([{spawn, Pid, _}|Tail], State0=#state{procs=Procs}) ->
	ProcStart = erlang:system_time(nanosecond),
	State = State0#state{procs=Procs#{Pid => #{spawn => ProcStart}}},
	fold(Tail, State);
fold([{response, _, _, Body}|Tail], State) ->
	Resp = erlang:system_time(nanosecond),
	fold(Tail, State#state{
		resp_start=Resp,
		resp_end=Resp,
		resp_body_length=resp_body_length(Body)
	});
fold([{headers, _, _}|Tail], State) ->
	RespStart = erlang:system_time(nanosecond),
	fold(Tail, State#state{resp_start=RespStart});
fold([{data, nofin, Data}|Tail], State=#state{resp_body_length=RespBodyLen}) ->
	fold(Tail, State#state{
		resp_body_length=RespBodyLen + resp_body_length(Data)
	});
fold([{data, fin, Data}|Tail], State=#state{resp_body_length=RespBodyLen}) ->
	RespEnd = erlang:system_time(nanosecond),
	fold(Tail, State#state{
		resp_end=RespEnd,
		resp_body_length=RespBodyLen + resp_body_length(Data)
	});
fold([_|Tail], State) ->
	fold(Tail, State).

resp_body_length({sendfile, _, Len, _}) ->
	Len;
resp_body_length(Data) ->
	iolist_size(Data).

-spec terminate(cowboy_stream:streamid(), cowboy_stream:reason(), #state{}) -> any().
terminate(StreamID, Reason, #state{next=Next, callback=Fun, ref=Ref,
		req_start=ReqStart, req_end=ReqEnd, req_body_start=ReqBodyStart,
		req_body_end=ReqBodyEnd, resp_start=RespStart, resp_end=RespEnd,
		procs=Procs, req_body_length=ReqBodyLen, resp_body_length=RespBodyLen}) ->
	Res = cowboy_stream:terminate(StreamID, Reason, Next),
	ReqEnd = erlang:system_time(nanosecond),
	Metrics = #{
		ref => Ref,
		pid => self(),
		streamid => StreamID,

		%% @todo Req (filtered)
		%% @todo Resp (filtered?)
		%% @todo Handler...

		reason => Reason,
		req_start => ReqStart,
		req_end => ReqEnd,
		req_body_start => ReqBodyStart,
		req_body_end => ReqBodyEnd,
		resp_start => RespStart,
		resp_end => RespEnd,
		procs => Procs,
		req_body_length => ReqBodyLen,
		resp_body_length => RespBodyLen
	},
	Fun(Metrics),
	Res.

-spec early_error(cowboy_stream:streamid(), cowboy_stream:reason(),
	cowboy_stream:partial_req(), Resp, cowboy:opts()) -> Resp
	when Resp::cowboy_stream:resp_command().
early_error(StreamID, Reason, PartialReq=#{ref := Ref}, Resp0, Opts=#{metrics_callback := Fun}) ->
	Time = erlang:system_time(nanosecond),
	Resp = {response, _, _, Body}
		= cowboy_stream:early_error(StreamID, Reason, PartialReq, Resp0, Opts),
	%% As far as metrics go we are limited in what we can provide
	%% in this case.
	Metrics = #{
		ref => Ref,
		pid => self(),
		streamid => StreamID,
		reason => Reason,

		%% @todo PartialReq
		%% @todo Resp status/headers (filtered?)

		early_error_time => Time,
		resp_body_length => resp_body_length(Body)
	},
	Fun(Metrics),
	Resp.