Merge pull request #151 from esl/configurable-env-parser

Configurable env parser

Makefile

@@ -1,4 +1,4 @@
-.PHONY: default rel compile clean ct test integration_test dialyzer xref console
+.PHONY: default rel compile clean ct test integration_test dialyzer xref console lint
 
 REBAR = rebar3
 
@@ -25,7 +25,10 @@ ct:
 	## eunit and ct commands always add a test profile to the run
 	$(REBAR) ct --verbose $(SUITE_OPTS)
 
-test: compile xref dialyzer ct
+lint:
+	$(REBAR) as elvis lint
+
+test: compile xref dialyzer ct lint
 
 integration_test:
 	./integration_test/stop_demo_cluster.sh

elvis.config

@@ -0,0 +1,28 @@
+[{elvis, [
+    {config, [
+        #{dirs => ["src", "src/*", "scenarios"],
+          filter => "*.erl",
+          ruleset => erl_files,
+          rules => [
+              {elvis_style, invalid_dynamic_call, #{ignore => [amoc_user]}},
+              {elvis_style, export_used_types, disable},
+              {elvis_style, no_throw, #{ignore => [{amoc_config, get, 2}] }},
+              {elvis_text_style, line_length, #{skip_comments => whole_line }},
+              {elvis_style, no_block_expressions, disable}
+          ]},
+        #{dirs => ["test"],
+          filter => "*.erl",
+          ruleset => erl_files,
+          rules => [
+              {elvis_style, function_naming_convention, #{regex => "^[a-z]([a-z0-9]*_?)*$"}},
+              {elvis_style, atom_naming_convention, #{regex => "^[a-z]([a-z0-9]*_?)*(_SUITE)?$"}},
+              {elvis_style, dont_repeat_yourself, #{min_complexity => 50}},
+              {elvis_style, no_debug_call, disable},
+              {elvis_style, no_throw, disable},
+              {elvis_style, no_import, disable}
+          ]},
+        #{dirs => ["."],
+          filter => "rebar.config",
+          ruleset => rebar_config}
+    ]}
+]}].

rebar.config

@@ -14,6 +14,7 @@
             {fusco, "0.1.1"}
         ]}
     ]},
+    {elvis, [{plugins, [{rebar3_lint, "3.0.1"}]}]},
     {demo, [
         {erl_opts, [debug_info, {src_dirs, ["src", "scenarios"]}]},
         {relx, [

src/amoc_config/amoc_config.hrl

@@ -46,8 +46,9 @@
 -type maybe_module_config() :: {ok, [module_parameter()]} | error().
 
 -type one_of() :: [value(), ...].
--type verification_method() :: none | one_of() | {module(), atom(), 1}.
--type update_method() :: read_only | none | {module(), atom(), 2}.
+-type mfa(Arity) :: {module(), atom(), Arity}.
+-type verification_method() :: none | one_of() | mfa(1) | verification_fun().
+-type update_method() :: read_only | none | mfa(2) | update_fun().
 
 -type module_attribute() :: #{ name := name(),
                                description := string(),

src/amoc_config/amoc_config_attributes.erl

@@ -127,8 +127,8 @@ make_module_parameter(#{name := Name, description := Description, default_value
     maybe_verification_fun() | not_exported | invalid_method.
 verification_fn(none) ->
     fun ?MODULE:none/1;
-verification_fn([_ | _] = OneOF) ->
-    one_of_fun(OneOF);
+verification_fn([_ | _] = OneOf) ->
+    one_of_fun(OneOf);
 verification_fn(Fun) when is_function(Fun, 1) ->
     is_exported(Fun);
 verification_fn({Module, Function, 1}) ->

src/amoc_config/amoc_config_env.erl

@@ -9,10 +9,14 @@
 %%==============================================================================
 -module(amoc_config_env).
 
--export([get/1, get/2, parse_value/1, format/2]).
+-export([get/1, get/2]).
 
 -include_lib("kernel/include/logger.hrl").
 
+-define(DEFAULT_PARSER_MODULE, amoc_config_parser).
+
+-callback(parse_value(string()) -> {ok, amoc_config:value()} | {error, any()}).
+
 %% ------------------------------------------------------------------
 %% API
 %% ------------------------------------------------------------------
@@ -24,24 +28,6 @@ get(Name) ->
 get(Name, Default) when is_atom(Name) ->
     get_os_env(Name, Default).
 
--spec parse_value(string() | binary()) -> {ok, amoc_config:value()} | {error, any()}.
-parse_value(Binary) when is_binary(Binary) ->
-    parse_value(binary_to_list(Binary));
-parse_value(String) when is_list(String) ->
-    try
-        {ok, Tokens, _} = erl_scan:string(String ++ "."),
-        {ok, _} = erl_parse:parse_term(Tokens)
-    catch
-        _:E -> {error, E}
-    end.
-
--spec format(any(), binary) -> binary();
-            (any(), string)  -> string().
-format(Value, binary) ->
-    list_to_binary(format(Value, string));
-format(Value, string) ->
-    lists:flatten(io_lib:format("~tp", [Value])).
-
 %% ------------------------------------------------------------------
 %% Internal Function Definitions
 %% ------------------------------------------------------------------
@@ -51,8 +37,13 @@ get_os_env(Name, Default) ->
     Value = os:getenv(EnvName),
     case parse_value(Value, Default) of
         {ok, Term} -> Term;
-        {error, _} ->
-            ?LOG_ERROR("cannot parse $~p value \"~p\", using default one", [EnvName, Value]),
+        {error, Error} ->
+            ?LOG_ERROR("cannot parse environment variable, using default value.~n"
+                          "  parsing error:  '~p'~n"
+                          "  variable name:  '$~s'~n"
+                          "  variable value: '~s'~n"
+                          "  default value:  '~p'~n",
+                       [Error, EnvName, Value, Default]),
             Default
     end.
 
@@ -64,4 +55,13 @@ os_env_name(Name) when is_atom(Name) ->
 parse_value(false, Default) -> {ok, Default};
 parse_value("", Default)    -> {ok, Default};
 parse_value(String, _) ->
-    parse_value(String).
+    App = application:get_application(?MODULE),
+    Mod = application:get_env(App, config_parser_mod, ?DEFAULT_PARSER_MODULE),
+    try Mod:parse_value(String) of
+        {ok, Value} -> {ok, Value};
+        {error, Error} -> {error, Error};
+        InvalidRetValue -> {error, {parser_returned_invalid_value, InvalidRetValue}}
+    catch
+        Class:Error:Stacktrace ->
+            {error, {parser_crashed, {Class, Error, Stacktrace}}}
+    end.

src/amoc_config/amoc_config_parser.erl

@@ -0,0 +1,40 @@
+%%==============================================================================
+%% Copyright 2023 Erlang Solutions Ltd.
+%% Licensed under the Apache License, Version 2.0 (see LICENSE file)
+%%==============================================================================
+%% This module implements the default parser for the amoc_config_env module
+%%==============================================================================
+-module(amoc_config_parser).
+-behaviour(amoc_config_env).
+
+-export([parse_value/1]).
+
+-ifdef(TEST).
+%% exported for testing only
+-export([format/2]).
+-else.
+-ignore_xref([format/2]).
+-dialyzer({nowarn_function, [format/2]}).
+-endif.
+
+%% ------------------------------------------------------------------
+%% API
+%% ------------------------------------------------------------------
+
+-spec parse_value(string() | binary()) -> {ok, amoc_config:value()} | {error, any()}.
+parse_value(Binary) when is_binary(Binary) ->
+    parse_value(binary_to_list(Binary));
+parse_value(String) when is_list(String) ->
+    try
+        {ok, Tokens, _} = erl_scan:string(String ++ "."),
+        {ok, _} = erl_parse:parse_term(Tokens)
+    catch
+        _:E -> {error, E}
+    end.
+
+-spec format(any(), binary) -> binary();
+            (any(), string)  -> string().
+format(Value, binary) ->
+    list_to_binary(format(Value, string));
+format(Value, string) ->
+    lists:flatten(io_lib:format("~tp", [Value])).

src/amoc_config/amoc_config_scenario.erl

@@ -39,7 +39,7 @@ update_settings(Settings) ->
         {fun verify_settings/2, [Settings]},
         {fun filter_configuration/2, [Settings]},
         {fun amoc_config_verification:process_scenario_config/2, [Settings]},
-        {fun store_scenario_config/1, []}],
+        {fun store_scenario_config_and_run_update_functions/1, []}],
     case amoc_config_utils:pipeline(PipelineActions, ok) of
         ok -> ok;
         {error, _, _} = Error -> Error;
@@ -129,12 +129,17 @@ filter_configuration(Config, Settings) ->
     FilteredConfig = [lists:keyfind(Name, KeyPos, Config) || Name <- Keys],
     case [{N, M} || #module_parameter{name = N, mod = M,
                                       update_fn = read_only} <- FilteredConfig] of
-        [] -> {ok, FilteredConfig};
+        [] ->
+            %% filter out unchanged parameters
+            ChangedParameters =
+                [P || #module_parameter{name = N, value = V} = P <- FilteredConfig,
+                      V =/= proplists:get_value(N, Settings)],
+            {ok, ChangedParameters};
         ReadOnlyParameters ->
             {error, readonly_parameters, ReadOnlyParameters}
     end.
 
-store_scenario_config(Config) ->
+store_scenario_config_and_run_update_functions(Config) ->
     amoc_config_utils:store_scenario_config(Config),
     [spawn(fun() -> apply(Fn, [Name, Value]) end)
      || #module_parameter{name = Name, value = Value, update_fn = Fn} <- Config],

src/amoc_controller.erl

@@ -95,10 +95,10 @@ update_settings(Settings) ->
 
 -spec add_users(amoc_scenario:user_id(), amoc_scenario:user_id()) ->
     ok | {error, term()}.
-add_users(StartId, EndID) ->
-    telemetry:execute([amoc, controller, users], #{count => EndID - StartId + 1}, #{type => add}),
+add_users(StartId, EndId) ->
+    telemetry:execute([amoc, controller, users], #{count => EndId - StartId + 1}, #{type => add}),
     %% adding the exact range of the users
-    gen_server:call(?SERVER, {add, StartId, EndID}).
+    gen_server:call(?SERVER, {add, StartId, EndId}).
 
 -spec remove_users(user_count(), boolean()) -> {ok, user_count()}.
 remove_users(Count, ForceRemove) ->
@@ -145,8 +145,8 @@ handle_call(stop_scenario, _From, State) ->
 handle_call({update_settings, Settings}, _From, State) ->
     RetValue = handle_update_settings(Settings, State),
     {reply, RetValue, State};
-handle_call({add, StartId, EndID}, _From, State) ->
-    {RetValue, NewState} = handle_add(StartId, EndID, State),
+handle_call({add, StartId, EndId}, _From, State) ->
+    {RetValue, NewState} = handle_add(StartId, EndId, State),
     {reply, RetValue, NewState};
 handle_call({remove, Count, ForceRemove}, _From, State) ->
     RetValue = handle_remove(Count, ForceRemove, State),
@@ -296,13 +296,13 @@ init_scenario(Scenario, Settings) ->
         {error, Type, Reason} -> {error, {Type, Reason}}
     end.
 
--spec maybe_start_timer(timer:tref()|undefined) -> timer:tref().
+-spec maybe_start_timer(timer:tref() | undefined) -> timer:tref().
 maybe_start_timer(undefined) ->
     {ok, TRef} = timer:send_interval(interarrival(), start_user),
     TRef;
 maybe_start_timer(TRef) -> TRef.
 
--spec maybe_stop_timer(timer:tref()|undefined) -> undefined.
+-spec maybe_stop_timer(timer:tref() | undefined) -> undefined.
 maybe_stop_timer(undefined) ->
     undefined;
 maybe_stop_timer(TRef) ->

src/amoc_coordinator/amoc_coordinator.erl

@@ -86,7 +86,8 @@ start(Name, CoordinationPlan, Timeout) when ?IS_TIMEOUT(Timeout) ->
             %% end, we need to add them first.
             AllItemsHandlers = lists:reverse([Item || {all, _} = Item <- Plan]),
             [gen_event:add_handler(Name, ?MODULE, {Name, Item}) || Item <- AllItemsHandlers],
-            [gen_event:add_handler(Name, ?MODULE, {Name, Item}) || {N, _} = Item <- Plan, is_integer(N)],
+            [gen_event:add_handler(Name, ?MODULE, {Name, Item}) || {N, _} = Item <- Plan,
+                                                                   is_integer(N)],
             gen_event:add_handler(Name, ?MODULE, {timeout, Name, Timeout}),
             ok;
         {error, _} -> error

src/amoc_distribution/amoc_cluster.erl

@@ -168,9 +168,11 @@ connect_nodes(Node, Nodes) ->
 set_master_node(Node, Action) ->
     case gen_server:call({?SERVER, Node}, {set_master_node, node()}) of
         ok ->
-            case catch apply(Action, [Node]) of
+            try apply(Action, [Node]) of
                 ok -> gen_server:cast(?SERVER, {add_slave, Node});
                 RetValue -> {error, {invalid_action_ret_value, RetValue}}
+            catch
+                C:E:S -> {error, {invalid_action_ret_value, {C, E, S}}}
             end;
         Error -> Error
     end.

src/amoc_distribution/amoc_dist.erl

@@ -16,7 +16,7 @@
 
 -compile({no_auto_import, [ceil/1]}).
 
--type cluster_state():: idle | running | stopped.
+-type cluster_state() :: idle | running | stopped.
 %% ------------------------------------------------------------------
 %% API
 %% ------------------------------------------------------------------

src/amoc_throttle/amoc_throttle.erl

@@ -114,7 +114,7 @@ change_rate_gradually(Name, LowRate, HighRate, RateInterval, StepInterval, NoOfS
 %%        destroy Async runner
 %%  '''
 %% for the local execution, req/exec rates are increased only by throttle process.
--spec run(name(), fun(()-> any())) -> ok | {error, any()}.
+-spec run(name(), fun(() -> any())) -> ok | {error, any()}.
 run(Name, Fn) ->
     amoc_throttle_controller:run(Name, Fn).
 

src/amoc_throttle/amoc_throttle_controller.erl

@@ -55,7 +55,7 @@ start_link() ->
 ensure_throttle_processes_started(Name, Interval, Rate, NoOfProcesses) ->
     gen_server:call(?MASTER_SERVER, {start_processes, Name, Interval, Rate, NoOfProcesses}).
 
--spec run(name(), fun(()-> any())) -> ok | {error, any()}.
+-spec run(name(), fun(() -> any())) -> ok | {error, any()}.
 run(Name, Fn) ->
     case get_throttle_process(Name) of
         {ok, Pid} ->
@@ -239,7 +239,7 @@ rate_per_minute(Rate, Interval) ->
 start_processes(Name, Rate, Interval, NoOfProcesses) ->
     % Master metrics
     RatePerMinute = rate_per_minute(Rate, Interval),
-    telemetry:execute([amoc, throttle, rate], #{rate => RatePerMinute}, #{name => Name}),
+    report_rate(Name, RatePerMinute),
     RealNoOfProcesses = min(Rate, NoOfProcesses),
     start_throttle_processes(Name, Interval, Rate, RealNoOfProcesses),
     RealNoOfProcesses.
@@ -273,7 +273,7 @@ do_change_rate(Name, Rate, Interval) ->
         [] -> {error, no_processes_in_group};
         List when is_list(List) ->
             RatePerMinute = rate_per_minute(Rate, Interval),
-            telemetry:execute([amoc, throttle, rate], #{rate => RatePerMinute}, #{name => Name}),
+            report_rate(Name, RatePerMinute),
             update_throttle_processes(List, Interval, Rate, length(List)),
             {ok, RatePerMinute}
     end.
@@ -317,3 +317,6 @@ run_cmd(Pid, pause) ->
     amoc_throttle_process:pause(Pid);
 run_cmd(Pid, resume) ->
     amoc_throttle_process:resume(Pid).
+
+report_rate(Name, RatePerMinute) ->
+    telemetry:execute([amoc, throttle, rate], #{rate => RatePerMinute}, #{name => Name}).

test/amoc_config_attributes_SUITE.erl

@@ -32,6 +32,8 @@
     #{name => var7, description => "var7", default_value => def7,
       verification => none, update => {?MODULE, update_value, 2}}
 ]).
+-required_variable(#{name => var7b, description => "var7b", default_value => def7,
+                     verification => none, update => fun ?MODULE:update_value/2}).
 
 %% verification functions
 -export([verify_value/1]).
@@ -66,7 +68,9 @@ get_module_attributes(_) ->
         #{name => var6, description => "var6", default_value => def6,
           verification => none, update => none},
         #{name => var7, description => "var7", default_value => def7,
-          verification => none, update => {?MODULE, update_value, 2}}],
+          verification => none, update => {?MODULE, update_value, 2}},
+        #{name => var7b, description => "var7b", default_value => def7,
+          verification => none, update => fun ?MODULE:update_value/2}],
     ?assertEqual(ExpectedResult, Result).
 
 get_module_configuration(_) ->
@@ -95,6 +99,8 @@ get_module_configuration(_) ->
          #module_parameter{name = var6, mod = ?MODULE, value = def6,
                            verification_fn = VerificationNone, update_fn = UpdateNone},
          #module_parameter{name = var7, mod = ?MODULE, value = def7,
+                           verification_fn = VerificationNone, update_fn = UpdateValueFN},
+         #module_parameter{name = var7b, mod = ?MODULE, value = def7,
                            verification_fn = VerificationNone, update_fn = UpdateValueFN}],
         Config).
 
@@ -115,8 +121,10 @@ errors_reporting(_) ->
                       update => fun invalid_module:not_exported_function/2},
     InvalidParam7 = #{name => invalid_var7, description => "var7",
                       update => fun update_value/2}, %% local function
-    Attributes = [InvalidParam0, InvalidParam1, InvalidParam2, ValidParam3,
-                  InvalidParam4, InvalidParam4b, InvalidParam5, InvalidParam6, InvalidParam7],
+    InvalidParam7b = #{name => invalid_var7, description => "var7b",
+                      update => fun update_value/2}, %% local function
+    Attributes = [InvalidParam0, InvalidParam1, InvalidParam2, ValidParam3, InvalidParam4,
+                  InvalidParam4b, InvalidParam5, InvalidParam6, InvalidParam7, InvalidParam7b],
     {error, invalid_attribute_format, Reason} =
         amoc_config_attributes:process_module_attributes(?MODULE, Attributes),
     ?assertEqual(
@@ -127,7 +135,8 @@ errors_reporting(_) ->
          {InvalidParam4b, verification_method_not_exported},
          {InvalidParam5, invalid_update_method},
          {InvalidParam6, update_method_not_exported},
-         {InvalidParam7, update_method_not_exported}],
+         {InvalidParam7, update_method_not_exported},
+         {InvalidParam7b, update_method_not_exported}],
         Reason).
 
 one_of_function(_) ->