valitydev/bouncer
14
0
Fork 0
mirror of https://github.com/valitydev/bouncer.git synced 2024-11-06 02:15:18 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

MSPF-641: Relax fsync-every-write audit log handler policy (#20)

Browse Source 
* Make fsync repeat interval configurable as usual.

* Dig into logger internals to sync w/ handler reliably

* Upgrade to opa 0.26.0-debug service image in tests

* Bump to rbkmoney/image-build-erlang@61a001b
This commit is contained in:
Andrew Mayorov 2021-02-18 11:24:50 +03:00 committed by GitHub
parent 2e575b0654
commit 4b86d8e48f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 251 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@ -16,7 +16,8 @@ SERVICE_IMAGE_PUSH_TAG ?= $(SERVICE_IMAGE_TAG)
BASE_IMAGE_NAME := service-erlang
BASE_IMAGE_TAG := 688cee70c0eb6540709fe35b816c81a90dc542ea
BUILD_IMAGE_TAG := 917afcdd0c0a07bf4155d597bbba72e962e1a34a
BUILD_IMAGE_NAME := build-erlang
BUILD_IMAGE_TAG := 61a001bbb48128895735a3ac35b0858484fdb2eb
CALL_ANYWHERE := \
submodules \
all compile xref lint format check_format dialyze cover release clean distclean

4
README.md
View File

@ -7,3 +7,7 @@ Primary [Arbiter](https://github.com/rbkmoney/bouncer-proto/blob/97dcad6f/proto/
In a nutshell this service maps incoming contexts into [OPA input documents](https://www.openpolicyagent.org/docs/latest/philosophy/#the-opa-document-model) and asks OPA to compute a judgement allowing, restricting or forbidding actions under given input context.
From the service's point of view a **ruleset id** is a path to OPA document that define a subdocument named `judgement` with a rudimentary schema. See https://github.com/rbkmoney/bouncer-policies#authoring for more detailed information.
## Things to keep in mind
* When upgrading service to a newer Erlang OTP release please be aware of potential breakages in `bouncer_audit_log` not strictly related to usual API deprecations and removals which is an essintial part of a release. That is because this module relies on some OTP libraries' interal implementation details. See module-level notes for additional details.

2
docker-compose.sh
View File

@ -1,6 +1,6 @@
#!/bin/bash
OPA_VSN="0.23.2-debug"
OPA_VSN="0.26.0-debug"
OPA_IMAGE="openpolicyagent/opa:${OPA_VSN}"
POLICY_DIR="/var/opa/policies"

6
rebar.config
View File

@ -67,10 +67,14 @@
filter => "*.erl",
ruleset => erl_files,
rules => [
% Too opionated
{elvis_style, state_record_and_type, disable},
{elvis_style, invalid_dynamic_call, #{
ignore => [
% Uses thrift reflection through `struct_info/1`.
bouncer_thrift
bouncer_thrift,
% Implements parts of logger duties, including message formatting.
bouncer_audit_log
]
}}
]

23
rebar.lock
View File

@ -1,4 +1,4 @@
{"1.1.0",
{"1.2.0",
[{<<"bear">>,{pkg,<<"bear">>,<<"0.8.7">>},2},
{<<"bouncer_proto">>,
{git,"git@github.com:rbkmoney/bouncer-proto.git",
@ -92,5 +92,24 @@
{<<"recon">>, <<"430FFA60685AC1EFDFB1FE4C97B8767C92D0D92E6E7C3E8621559BA77598678A">>},
{<<"rfc3339">>, <<"1552DF616ACA368D982E9F085A0E933B6688A3F4938A671798978EC2C0C58730">>},
{<<"ssl_verify_fun">>, <<"6EAF7AD16CB568BB01753DBBD7A95FF8B91C7979482B95F38443FE2C8852A79B">>},
{<<"unicode_util_compat">>, <<"D869E4C68901DD9531385BB0C8C40444EBF624E60B6962D95952775CAC5E90CD">>}]}
{<<"unicode_util_compat">>, <<"D869E4C68901DD9531385BB0C8C40444EBF624E60B6962D95952775CAC5E90CD">>}]},
{pkg_hash_ext,[
{<<"bear">>, <<"534217DCE6A719D59E54FB0EB7A367900DBFC5F85757E8C1F94269DF383F6D9B">>},
{<<"cache">>, <<"3E7D6706DE5DF76C4D71C895B4BE62B01C3DE6EDB63197035E465C3BCE63F19B">>},
{<<"certifi">>, <<"805ABD97539CAF89EC6D4732C91E62BA9DA0CDA51AC462380BBD28EE697A8C42">>},
{<<"cowboy">>, <<"4643E4FBA74AC96D4D152C75803DE6FAD0B3FA5DF354C71AFDD6CBEEB15FAC8A">>},
{<<"cowlib">>, <<"E4175DC240A70D996156160891E1C62238EDE1729E45740BDD38064DAD476170">>},
{<<"gproc">>, <<"580ADAFA56463B75263EF5A5DF4C86AF321F68694E7786CB057FD805D1E2A7DE">>},
{<<"hackney">>, <<"E0100F8EF7D1124222C11AD362C857D3DF7CB5F4204054F9F0F4A728666591FC">>},
{<<"idna">>, <<"4BDD305EB64E18B0273864920695CB18D7A2021F31A11B9C5FBCD9A253F936E2">>},
{<<"jesse">>, <<"38D9C4743F97F073D0486CF6626AB370C842D94EE207BB3574052845979A7C0D">>},
{<<"jsx">>, <<"37BECA0435F5CA8A2F45F76A46211E76418FBEF80C36F0361C249FC75059DC6D">>},
{<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>},
{<<"mimerl">>, <<"F278585650AA581986264638EBF698F8BB19DF297F66AD91B18910DFC6E19323">>},
{<<"parse_trans">>, <<"17EF63ABDE837AD30680EA7F857DD9E7CED9476CDD7B0394432AF4BFC241B960">>},
{<<"ranch">>, <<"451D8527787DF716D99DC36162FCA05934915DB0B6141BBDAC2EA8D3C7AFC7D7">>},
{<<"recon">>, <<"5721C6B6D50122D8F68CCCAC712CAA1231F97894BAB779EFF5FF0F886CB44648">>},
{<<"rfc3339">>, <<"986D7F9BAC6891AA4D5051690058DE4E623245620BBEADA7F239F85C4DF8F23C">>},
{<<"ssl_verify_fun">>, <<"13104D7897E38ED7F044C4DE953A6C28597D1C952075EB2E328BC6D6F2BFC496">>},
{<<"unicode_util_compat">>, <<"1D1848C40487CDB0B30E8ED975E34E025860C02E419CB615D255849F3427439D">>}]}
].

33
src/bouncer.erl
View File

@ -5,7 +5,6 @@
-behaviour(application).
-export([start/2]).
-export([prep_stop/1]).
-export([stop/1]).
%% Supervisor callbacks
@ -20,14 +19,6 @@
start(_StartType, _StartArgs) ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
-spec prep_stop(State) -> State.
prep_stop(State) ->
% NOTE
% We have to do it in this magic `prep_stop/1` here because for some inexplicable reason the
% usual `stop/1` callback doesn't get called in common_test runs.
ok = bouncer_audit_log:stop(genlib_app:env(?MODULE, audit, #{})),
State.
-spec stop(any()) -> ok.
stop(_State) ->
ok.
@ -36,11 +27,11 @@ stop(_State) ->
-spec init([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
init([]) ->
AuditPulse = bouncer_audit_log:init(genlib_app:env(?MODULE, audit, #{})),
{AuditSpecs, AuditPulse} = get_audit_specs(),
ServiceOpts = genlib_app:env(?MODULE, services, #{}),
EventHandlers = genlib_app:env(?MODULE, woody_event_handlers, [woody_event_handler_default]),
Healthcheck = enable_health_logging(genlib_app:env(?MODULE, health_check, #{})),
ChildSpec = woody_server:child_spec(
WoodySpec = woody_server:child_spec(
?MODULE,
#{
ip => get_ip_address(),
@ -54,11 +45,21 @@ init([]) ->
additional_routes => [erl_health_handle:get_route(Healthcheck)]
}
),
{ok,
{
#{strategy => one_for_all, intensity => 6, period => 30},
[ChildSpec]
}}.
{ok, {
#{strategy => one_for_one, intensity => 10, period => 10},
AuditSpecs ++ [WoodySpec]
}}.
-spec get_audit_specs() -> {[supervisor:child_spec()], bouncer_arbiter_pulse:handlers()}.
get_audit_specs() ->
Opts = genlib_app:env(?MODULE, audit, #{}),
case maps:get(log, Opts, #{}) of
LogOpts = #{} ->
{ok, ChildSpec, Pulse} = bouncer_audit_log:child_spec(LogOpts),
{[ChildSpec], [Pulse]};
disable ->
{[], []}
end.
-spec get_ip_address() -> inet:ip_address().
get_ip_address() ->

299
src/bouncer_audit_log.erl
View File

@ -1,25 +1,25 @@
-module(bouncer_audit_log).
-export([init/1]).
-export([stop/1]).
-export([child_spec/1]).
-behaviour(bouncer_arbiter_pulse).
-export([handle_beat/3]).
-behaviour(gen_server).
-export([init/1]).
-export([handle_call/3]).
-export([handle_cast/2]).
-export([handle_info/2]).
-export([terminate/2]).
-define(HANDLER_ID, ?MODULE).
-define(DEFAULT_LOG_LEVEL, notice).
-define(DEFAULT_FLUSH_QLEN, 10000).
-define(LOG_DOMAIN, [audit]).
-define(DEFAULT_FORMATTER, {logger_formatter, #{single_line => true}}).
-type opts() :: #{
log => log_opts() | disabled
}.
% NOTE
% Keep in sync with `opts()`.
-define(OPTS, [log]).
-type log_opts() :: #{
% Which log level to use for audit events? Defaults to `notice`.
level => logger:level(),
backend => logger_backend_opts(),
@ -28,121 +28,83 @@
}.
% NOTE
% Keep in sync with `log_opts()`.
-define(LOG_OPTS, [level, backend, formatter]).
% Keep in sync with `opts()`.
-define(OPTS, [level, backend, formatter]).
-type logger_backend_opts() :: #{
% Where to log? Defaults to `standard_io`.
type => standard_io | standard_error | file,
% Log file location. No default, MUST be set if `type` is `file`.
file => file:filename(),
% How often to force fsync on log file, in ms? Defaults to 1000.
% http://erlang.org/doc/man/logger_std_h.html
filesync_repeat_interval => pos_integer() | no_repeat,
max_no_bytes => pos_integer() | infinity,
max_no_files => non_neg_integer(),
% Maximum number of events to queue for writing. Defaults to 10000.
% http://erlang.org/doc/apps/kernel/logger_chapter.html#message-queue-length
flush_qlen => non_neg_integer()
max_no_files => non_neg_integer()
}.
% NOTE
% Keep in sync with `logger_backend_opts()`.
-define(LOGGER_BACKEND_OPTS, [type, file, max_no_bytes, max_no_files, flush_qlen]).
-opaque st() :: {logger:level(), _Formatter :: {module(), logger:formatter_config()}}.
-export_type([opts/0]).
-export_type([st/0]).
%%
-type st() ::
{log, logger:level()}.
-spec init(opts()) -> bouncer_arbiter_pulse:handlers(st()).
init(Opts) ->
-spec child_spec(opts()) -> {ok, supervisor:child_spec(), bouncer_arbiter_pulse:handler(st())}.
child_spec(Opts) ->
_ = assert_strict_opts(?OPTS, Opts),
init_log_handler(maps:get(log, Opts, #{})).
init_log_handler(LogOpts = #{}) ->
_ = assert_strict_opts(?LOG_OPTS, LogOpts),
Level = validate_log_level(maps:get(level, LogOpts, ?DEFAULT_LOG_LEVEL)),
BackendConfig = mk_logger_backend_config(maps:get(backend, LogOpts, #{})),
HandlerConfig0 = maps:with([formatter], LogOpts),
HandlerConfig1 = HandlerConfig0#{
Level = validate_log_level(maps:get(level, Opts, ?DEFAULT_LOG_LEVEL)),
Formatter = maps:get(formatter, Opts, ?DEFAULT_FORMATTER),
BackendConfig = mk_logger_backend_config(maps:get(backend, Opts, #{})),
Config = #{
id => ?HANDLER_ID,
module => logger_std_h,
config => BackendConfig,
% NOTE
% This two options together ensure that _only_ audit logs will flow through to the backend.
filters => [{domain, {fun logger_filters:domain/2, {log, sub, ?LOG_DOMAIN}}}],
filter_default => stop
formatter => Formatter
},
ok = logger:add_handler(
?MODULE,
logger_std_h,
HandlerConfig1
),
% TODO
% Validate that global logger level doesn't suppress ours?
ok = log(Level, "audit log started", #{}),
[{?MODULE, {log, Level}}];
init_log_handler(disabled) ->
[].
ChildSpec = #{
id => ?MODULE,
start => {gen_server, start_link, [{local, ?HANDLER_ID}, ?MODULE, {Level, Config}, []]},
restart => permanent,
shutdown => 5000
},
Pulse = {?MODULE, {Level, Formatter}},
{ok, ChildSpec, Pulse}.
validate_log_level(Level) ->
eq = logger:compare_levels(Level, Level),
Level.
mk_logger_backend_config(BackendOpts) ->
_ = assert_strict_opts(?LOGGER_BACKEND_OPTS, BackendOpts),
Type = validate_log_type(maps:get(type, BackendOpts, standard_io)),
mk_logger_backend_config(Type, BackendOpts).
% NOTE
% Those are options only relevant for `logger_h_common` module, see its implementation for
% details.
Common = [filesync_repeat_interval],
CommonOpts = maps:with(Common, BackendOpts),
{ok, BackendConfig} = logger_std_h:check_config(
?HANDLER_ID,
set,
undefined,
tune_backend_config(maps:without(Common, BackendOpts))
),
maps:merge(
CommonOpts,
BackendConfig
).
validate_log_type(Type) when Type == standard_io; Type == standard_error; Type == file ->
Type;
validate_log_type(Type) ->
erlang:error(badarg, [Type]).
mk_logger_backend_config(file = Type, Opts) ->
Defaults = get_default_backend_config(Type, Opts),
Filename = maps:get(file, Opts),
Config0 = maps:with([max_no_bytes, max_no_files], Opts),
Config = maps:merge(Defaults, Config0),
Config#{
type => Type,
file => Filename
};
mk_logger_backend_config(Type, Opts) ->
Defaults = get_default_backend_config(Type, Opts),
Defaults#{
type => Type
}.
get_default_backend_config(file, Opts) ->
tune_backend_config(Opts = #{type := file}) ->
_ = maps:get(file, Opts),
% NOTE
% All those options chosen to push message loss probability as close to zero as possible.
% Zero doesn't seem reachable with standard logger infrastructure because of various safeguards
% around unexpected backend and formatter errors.
Config = get_default_backend_config(Opts),
Config#{
% Protects against accidental write loss upon file rotation.
file_check => 0
};
get_default_backend_config(_Type, Opts) ->
get_default_backend_config(Opts).
get_default_backend_config(Opts) ->
FlushQLen = maps:get(flush_qlen, Opts, ?DEFAULT_FLUSH_QLEN),
#{
% No need to set it up here since we'll sync on EVERY write by ourself.
filesync_repeat_interval => no_repeat,
% http://erlang.org/doc/apps/kernel/logger_chapter.html#message-queue-length
sync_mode_qlen => 0,
drop_mode_qlen => FlushQLen,
flush_qlen => FlushQLen,
% http://erlang.org/doc/apps/kernel/logger_chapter.html#controlling-bursts-of-log-requests
burst_limit_enable => false,
% http://erlang.org/doc/apps/kernel/logger_chapter.html#terminating-an-overloaded-handler
overload_kill_enable => false
}.
maps:merge(
#{
% Protects against accidental write loss upon file rotation.
file_check => 0
},
Opts
);
tune_backend_config(Opts) ->
Opts.
assert_strict_opts(Ks, Opts) ->
case maps:without(Ks, Opts) of
@ -154,18 +116,65 @@ assert_strict_opts(Ks, Opts) ->
%%
-spec stop(opts()) -> ok.
stop(Opts = #{}) ->
stop_log_handler(maps:get(log, Opts, #{})).
-spec init({logger:level(), logger:handler_config()}) -> {ok, _State}.
init({Level, Config}) ->
% NOTE
% Here we're hijacking `logger_h_common` facility which drives log handler activities and
% periodic filesyncing. This gives us a thin wrapper supporting both logger handler API and
% an ability to make strictly synchronous writes which aren't really available through regular
% logger API. Note that we deliberately gave up on overload protection facilities (provided
% through `logger_olp`).
case logger_h_common:init(Config) of
{ok, State1} ->
Formatter = maps:get(formatter, Config),
StartEvent = mk_log_event(Level, "audit log started", #{}),
{ok, State2} = emit_log_sync(log_to_binary(StartEvent, Formatter), State1),
undefined = erlang:put(?MODULE, {Level, Formatter}),
{ok, State2};
Error ->
erlang:exit(Error)
end.
-spec stop_log_handler(log_opts()) -> ok.
stop_log_handler(LogOpts = #{}) ->
Level = maps:get(level, LogOpts, ?DEFAULT_LOG_LEVEL),
ok = log(Level, "audit log stopped", #{}),
_ = logger:remove_handler(?MODULE),
ok;
stop_log_handler(disabled) ->
ok.
-spec handle_call({?MODULE, binary()} | _Call, _From, State) -> {reply, ok | _Result, State}.
handle_call({?MODULE, Bin}, _From, State1) ->
{Result, State2} = emit_log_sync(Bin, State1),
{reply, Result, State2};
handle_call(Call, From, State) ->
logger_h_common:handle_call(Call, From, State).
-spec handle_cast(_Cast, State) -> {noreply, State}.
handle_cast(Cast, State) ->
logger_h_common:handle_cast(Cast, State).
-spec handle_info(_Info, State) -> {noreply, State}.
handle_info(Info, State) ->
logger_h_common:handle_info(Info, State).
-spec terminate(_Reason, _State) -> _.
terminate(Reason, State1) ->
{Level, Formatter} = erlang:get(?MODULE),
StopEvent = mk_log_event(Level, "audit log stopped", #{}),
{_, State2} = emit_log_sync(log_to_binary(StopEvent, Formatter), State1),
logger_h_common:terminate(Reason, State2).
% NOTE
% Be warned that this is IMPLEMENTATION DETAILS and is SUBJECT TO CHANGE!
% This code was adapted from `logger_h_common:handle_load/2` function found in kernel-7.2 app
% which is part of Erlang OTP 23.2.3 release.
% Please take care to update it when upgrading to newer Erlang OTP releases.
emit_log_sync(
Bin,
State = #{
id := Name,
module := Module,
handler_state := HandlerState
}
) ->
{Result, HS1} = Module:write(Name, sync, Bin, HandlerState),
{Result, State#{
handler_state => HS1,
last_op => write
}}.
%%
@ -173,27 +182,57 @@ stop_log_handler(disabled) ->
-type metadata() :: bouncer_arbiter_pulse:metadata().
-spec handle_beat(beat(), metadata(), st()) -> ok.
handle_beat(Beat, Metadata, {log, Level}) ->
log(
get_severity(Beat, Level),
get_message(Beat),
extract_metadata(Metadata, get_beat_metadata(Beat))
).
handle_beat(Beat, Metadata, {DefaultLevel, Formatter}) ->
case get_level(Beat, DefaultLevel) of
undefined ->
ok;
Level ->
log(
Level,
get_message(Beat),
extract_metadata(Metadata, get_beat_metadata(Beat)),
Formatter
)
end.
log(Severity, Message, Metadata) ->
DefaultMetadata = #{
type => audit,
domain => ?LOG_DOMAIN
},
% NOTE
% Matching on `ok` here is crucial. Logger may decide to flush the queue behind the scenes so
% we need to ensure it's not happening.
ok = logger:log(Severity, Message, maps:merge(Metadata, DefaultMetadata)),
ok = logger_std_h:filesync(?MODULE),
ok.
log(Level, Message, Metadata, Formatter) ->
Event = mk_log_event(Level, Message, Metadata),
ok = gen_server:call(?HANDLER_ID, {?MODULE, log_to_binary(Event, Formatter)}).
get_severity({judgement, started}, _Level) -> debug;
get_severity(_, Level) -> Level.
log_to_binary(Log, {Formatter, FormatterConfig}) ->
string_to_binary(Formatter:format(Log, FormatterConfig)).
string_to_binary(String) ->
case unicode:characters_to_binary(String) of
Binary when is_binary(Binary) ->
Binary;
Error ->
erlang:error(Error)
end.
mk_log_event(Level, Message, Metadata) ->
#{
level => Level,
msg => {Message, []},
meta => add_default_metadata(Metadata)
}.
add_default_metadata(Meta1) ->
Meta2 = Meta1#{type => audit},
Meta3 = maps:merge(get_process_metadata(), Meta2),
logger:add_default_metadata(Meta3).
get_process_metadata() ->
genlib:define(logger:get_process_metadata(), #{}).
get_level({judgement, started}, _Level) -> log_allowed(debug);
get_level(_, Level) -> Level.
log_allowed(Level) ->
case logger:allow(Level, ?MODULE) of
true -> Level;
false -> undefined
end.
get_message({judgement, started}) -> <<"judgement started">>;
get_message({judgement, {completed, _}}) -> <<"judgement completed">>;

59
test/bouncer_audit_tests_SUITE.erl
View File

@ -4,6 +4,7 @@
-include_lib("stdlib/include/assert.hrl").
-export([all/0]).
-export([groups/0]).
-export([init_per_suite/1]).
-export([end_per_suite/1]).
-export([init_per_testcase/2]).
@ -12,7 +13,7 @@
-export([invalid_config_fails_start/1]).
-export([unrecognized_config_fails_start/1]).
-export([write_error_fails_request/1]).
-export([write_queue_overload_fails_request/1]).
-export([write_queue_contention/1]).
-include_lib("bouncer_proto/include/bouncer_decisions_thrift.hrl").
@ -27,18 +28,21 @@
-define(OPA_ENDPOINT, {?OPA_HOST, 8181}).
-define(API_RULESET_ID, "service/authz/api").
-spec all() -> [atom()].
-spec all() -> [ct_suite:ct_test_def()].
all() ->
[
invalid_config_fails_start,
unrecognized_config_fails_start,
write_error_fails_request
% TODO
% This testcase is currently failing consistently.
% Turns out logger (as of Erlang/OTP 23 w/ kernel 7.0) doesn't tell callers when specific
% handler loses an event due to queue flushing, even though doc says it should have.
% Best bet is to max out `flush_qlen` option in the meantime.
% write_queue_overload_fails_request
{group, write_error_fails_request},
write_queue_contention
].
-spec groups() -> [ct_suite:ct_group_def()].
groups() ->
[
{write_error_fails_request, [], [
{testcase, write_error_fails_request, [{repeat, 10}]}
]}
].
-spec init_per_suite(config()) -> config().
@ -69,7 +73,7 @@ end_per_testcase(_Name, _C) ->
-spec invalid_config_fails_start(config()) -> ok.
-spec unrecognized_config_fails_start(config()) -> ok.
-spec write_error_fails_request(config()) -> ok.
-spec write_queue_overload_fails_request(config()) -> ok.
-spec write_queue_contention(config()) -> ok.
invalid_config_fails_start(C) ->
?assertError(
@ -104,15 +108,6 @@ invalid_config_fails_start(C) ->
).
unrecognized_config_fails_start(C) ->
?assertError(
{bouncer, {{{unrecognized_opts, #{blarg := _}}, _Stacktrace}, _}},
start_stop_bouncer(
[
{audit, #{blarg => blorg}}
],
C
)
),
?assertError(
{bouncer, {{{unrecognized_opts, #{blarg := _}}, _Stacktrace}, _}},
start_stop_bouncer(
@ -164,7 +159,10 @@ write_error_fails_request(C) ->
ok = file:delete(Filename),
ok = file:change_mode(Dirname, 8#555),
?assertError(
{woody_error, {external, result_unexpected, _}},
% NOTE
% The `_Reason` here may be either `result_unexpected` or `result_unknown`, depending
% on how fast application master is to stop the bouncer app.
{woody_error, {external, _Reason, _}},
call_judge(?API_RULESET_ID, ?CONTEXT(#{}), Client)
)
after
@ -172,16 +170,19 @@ write_error_fails_request(C) ->
stop_bouncer(C1)
end.
write_queue_overload_fails_request(C) ->
QLen = 10,
Concurrency = QLen * 10,
write_queue_contention(C) ->
Concurrency = 500,
Dirname = mk_temp_dir(?CONFIG(testcase, C)),
Filename = filename:join(Dirname, "audit.log"),
C1 = start_bouncer(
[
{audit, #{
log => #{
backend => #{type => file, file => Filename, flush_qlen => QLen},
backend => #{
type => file,
file => Filename,
filesync_repeat_interval => 0
},
formatter => {logger_logstash_formatter, #{single_line => true}}
}
}}
@ -199,9 +200,13 @@ write_queue_overload_fails_request(C) ->
try
{Succeeded, _Failed} = lists:partition(fun({R, _}) -> R == ok end, Results),
{ok, LogfileContents} = file:read_file(Filename),
% TODO kinda hacky
NumLogEvents = binary:matches(LogfileContents, <<"\"type\":\"audit\"">>),
?assertEqual(length(Succeeded), length(NumLogEvents))
LogfileLines = [string:trim(L) || L <- string:split(LogfileContents, "\n", all)],
LogfileEvents = [jsx:decode(L) || L <- LogfileLines, byte_size(L) > 0],
CompletedEvents = [
Event
|| Event = #{<<"judgement">> := #{<<"event">> := <<"completed">>}} <- LogfileEvents
],
?assertEqual(length(Succeeded), length(CompletedEvents))
after
rm_temp_dir(Dirname)
end.