12741272085

Committed 13 Jan 2025 05:33AM UTC coverage: 82.355%. First build

Build # 12741272085

Build Type

Pull #14475

github

Committed by

web-flow

Commit Message

Merge b56877530 into 0fc8025be

Pull Request Pull Request #14475: refactor(limiter): refactor and simplify the limiter

Run Details

199 of 249 new or added lines in 21 files covered. (79.92%)

57201 of 69457 relevant lines covered (82.35%)

15186.26 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.18

/apps/emqx/src/emqx_limiter/src/emqx_limiter_allocator_sup.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2021-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_limiter_allocator_sup).

-behaviour(supervisor).

%% API
-export([start_link/0, start/1, stop/1]).

%% Supervisor callbacks
-export([init/1]).

%%--------------------------------------------------------------------
%%  API functions
%%--------------------------------------------------------------------

%%--------------------------------------------------------------------
%% @doc
%% Starts the supervisor
%% @end
%%--------------------------------------------------------------------
-spec start_link() ->
    {ok, Pid :: pid()}
    | {error, {already_started, Pid :: pid()}}
    | {error, {shutdown, term()}}
    | {error, term()}
    | ignore.
start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

-spec start(emqx_limiter:zone()) -> _.
start(Zone) ->
    case has_limiter(Zone) of
        true ->
            Spec = make_child(Zone),
            supervisor:start_child(?MODULE, Spec);
        _ ->
            {error, <<"No Limiter">>}
    end.

stop(Zone) ->
    _ = supervisor:terminate_child(?MODULE, Zone),
    supervisor:delete_child(?MODULE, Zone).

%%--------------------------------------------------------------------
%%  Supervisor callbacks
%%--------------------------------------------------------------------

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Whenever a supervisor is started using supervisor:start_link/[2,3],
%% this function is called by the new process to find out about
%% restart strategy, maximum restart intensity, and child
%% specifications.
%% @end
%%--------------------------------------------------------------------
-spec init(Args :: term()) ->
    {ok, {SupFlags :: supervisor:sup_flags(), [ChildSpec :: supervisor:child_spec()]}}
    | ignore.
init([]) ->
    SupFlags = #{
        strategy => one_for_one,
        intensity => 10,
        period => 3600
    },
    {ok, {SupFlags, childs()}}.

%%--==================================================================
%%  Internal functions
%%--==================================================================
make_child(Zone) ->
    #{
        id => Zone,
        start => {emqx_limiter_allocator, start_link, [Zone]},
        restart => transient,
        shutdown => 5000,
        type => worker,
        modules => [emqx_limiter_allocator]
    }.

childs() ->
    Zones = maps:keys(emqx_config:get([zones])),
    lists:foldl(
        fun(Zone, Acc) ->
            case has_limiter(Zone) of
                true ->
                    [make_child(Zone) | Acc];
                _ ->
                    Acc
            end
        end,
        [make_child(emqx_limiter:internal_allocator())],
        Zones
    ).

has_limiter(Zone) ->
    case emqx_config:get_zone_conf(Zone, [mqtt, limiter], undefined) of
        undefined ->
            false;
        Cfg ->
            has_any_rate(Cfg)
    end.

has_any_rate(Cfg) ->
    Names = emqx_limiter_schema:mqtt_limiter_names(),
    lists:any(
        fun(Name) ->
            {ok, RateKey} = emqx_limiter:to_rate_key(Name),
            maps:get(RateKey, Cfg, infinity) =/= infinity
        end,
        Names
    ).

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2021-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(emqx_limiter_allocator_sup).
18
19	-behaviour(supervisor).
20
21	%% API
22	-export([start_link/0, start/1, stop/1]).
23
24	%% Supervisor callbacks
25	-export([init/1]).
26
27	%%--------------------------------------------------------------------
28	%% API functions
29	%%--------------------------------------------------------------------
30
31	%%--------------------------------------------------------------------
32	%% @doc
33	%% Starts the supervisor
34	%% @end
35	%%--------------------------------------------------------------------
36	-spec start_link() ->
37	{ok, Pid :: pid()}
38	\| {error, {already_started, Pid :: pid()}}
39	\| {error, {shutdown, term()}}
40	\| {error, term()}
41	\| ignore.
42	start_link() ->
43	supervisor:start_link({local, ?MODULE}, ?MODULE, []).	1,179✔
44
45	-spec start(emqx_limiter:zone()) -> _.
46	start(Zone) ->
NEW 47	case has_limiter(Zone) of	×
48	true ->
NEW 49	Spec = make_child(Zone),	×
NEW 50	supervisor:start_child(?MODULE, Spec);	×
51	_ ->
NEW 52	{error, <<"No Limiter">>}	×
53	end.
54
55	stop(Zone) ->
NEW 56	_ = supervisor:terminate_child(?MODULE, Zone),	×
NEW 57	supervisor:delete_child(?MODULE, Zone).	×
58
59	%%--------------------------------------------------------------------
60	%% Supervisor callbacks
61	%%--------------------------------------------------------------------
62
63	%%--------------------------------------------------------------------
64	%% @private
65	%% @doc
66	%% Whenever a supervisor is started using supervisor:start_link/[2,3],
67	%% this function is called by the new process to find out about
68	%% restart strategy, maximum restart intensity, and child
69	%% specifications.
70	%% @end
71	%%--------------------------------------------------------------------
72	-spec init(Args :: term()) ->
73	{ok, {SupFlags :: supervisor:sup_flags(), [ChildSpec :: supervisor:child_spec()]}}
74	\| ignore.
75	init([]) ->
76	SupFlags = #{	1,179✔
77	strategy => one_for_one,
78	intensity => 10,
79	period => 3600
80	},
81	{ok, {SupFlags, childs()}}.	1,179✔
82
83	%%--==================================================================
84	%% Internal functions
85	%%--==================================================================
86	make_child(Zone) ->
87	#{	1,179✔
88	id => Zone,
89	start => {emqx_limiter_allocator, start_link, [Zone]},
90	restart => transient,
91	shutdown => 5000,
92	type => worker,
93	modules => [emqx_limiter_allocator]
94	}.
95
96	childs() ->
97	Zones = maps:keys(emqx_config:get([zones])),	1,179✔
98	lists:foldl(	1,179✔
99	fun(Zone, Acc) ->
100	case has_limiter(Zone) of	1,181✔
101	true ->
NEW 102	[make_child(Zone) \| Acc];	×
103	_ ->
104	Acc	1,181✔
105	end
106	end,
107	[make_child(emqx_limiter:internal_allocator())],
108	Zones
109	).
110
111	has_limiter(Zone) ->
112	case emqx_config:get_zone_conf(Zone, [mqtt, limiter], undefined) of	1,181✔
113	undefined ->
114	false;	1,179✔
115	Cfg ->
116	has_any_rate(Cfg)	2✔
117	end.
118
119	has_any_rate(Cfg) ->
120	Names = emqx_limiter_schema:mqtt_limiter_names(),	2✔
121	lists:any(	2✔
122	fun(Name) ->
123	{ok, RateKey} = emqx_limiter:to_rate_key(Name),	6✔
124	maps:get(RateKey, Cfg, infinity) =/= infinity	6✔
125	end,
126	Names
127	).

emqx / emqx / 12741272085

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous