yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/release-57' into 20240724-sync-release-57

This commit is contained in:
Ivan Dyachkov 2024-07-24 10:28:00 +02:00
parent 9c0f1df8a3 c7a7658c7a
commit 606d829256
110 changed files with 1503 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx_auth/include/emqx_authn.hrl
View File

@ -28,7 +28,7 @@
-type authenticator_id() :: binary().
-define(AUTHN_RESOURCE_GROUP, <<"emqx_authn">>).
-define(AUTHN_RESOURCE_GROUP, <<"authn">>).
%% VAR_NS_CLIENT_ATTRS is added here because it can be initialized before authn.
%% NOTE: authn return may add more to (or even overwrite) client_attrs.

2
apps/emqx_auth/include/emqx_authz.hrl
View File

@ -156,7 +156,7 @@
count => 1
}).
-define(AUTHZ_RESOURCE_GROUP, <<"emqx_authz">>).
-define(AUTHZ_RESOURCE_GROUP, <<"authz">>).
-define(AUTHZ_FEATURES, [rich_actions]).

2
apps/emqx_auth/src/emqx_authz/emqx_authz_rule.erl
View File

@ -198,7 +198,7 @@ qos_from_opts(Opts) ->
)
end
catch
{bad_qos, QoS} ->
throw:{bad_qos, QoS} ->
throw(#{
reason => invalid_authorization_qos,
qos => QoS

57
apps/emqx_auth/src/emqx_authz/emqx_authz_utils.erl
View File

@ -16,6 +16,9 @@
-module(emqx_authz_utils).
-feature(maybe_expr, enable).
-include_lib("emqx/include/emqx_placeholder.hrl").
-include_lib("emqx_authz.hrl").
-include_lib("snabbkaffe/include/trace.hrl").
@ -28,7 +31,7 @@
remove_resource/1,
update_config/2,
vars_for_rule_query/2,
parse_rule_from_row/2
do_authorize/6
]).
-export([
@ -133,14 +136,18 @@ content_type(Headers) when is_list(Headers) ->
-define(RAW_RULE_KEYS, [<<"permission">>, <<"action">>, <<"topic">>, <<"qos">>, <<"retain">>]).
parse_rule_from_row(ColumnNames, Row) ->
RuleRaw = maps:with(?RAW_RULE_KEYS, maps:from_list(lists:zip(ColumnNames, to_list(Row)))),
case emqx_authz_rule_raw:parse_rule(RuleRaw) of
-spec parse_rule_from_row([binary()], [binary()] | map()) ->
{ok, emqx_authz_rule:rule()} | {error, term()}.
parse_rule_from_row(_ColumnNames, RuleMap = #{}) ->
case emqx_authz_rule_raw:parse_rule(RuleMap) of
{ok, {Permission, Action, Topics}} ->
emqx_authz_rule:compile({Permission, all, Action, Topics});
{ok, emqx_authz_rule:compile({Permission, all, Action, Topics})};
{error, Reason} ->
error(Reason)
end.
{error, Reason}
end;
parse_rule_from_row(ColumnNames, Row) ->
RuleMap = maps:with(?RAW_RULE_KEYS, maps:from_list(lists:zip(ColumnNames, to_list(Row)))),
parse_rule_from_row(ColumnNames, RuleMap).
vars_for_rule_query(Client, ?authz_action(PubSub, Qos) = Action) ->
Client#{
@ -157,3 +164,39 @@ to_list(Tuple) when is_tuple(Tuple) ->
tuple_to_list(Tuple);
to_list(List) when is_list(List) ->
List.
do_authorize(Type, Client, Action, Topic, ColumnNames, Row) ->
try
maybe
{ok, Rule} ?= parse_rule_from_row(ColumnNames, Row),
{matched, Permission} ?= emqx_authz_rule:match(Client, Action, Topic, Rule),
{matched, Permission}
else
nomatch ->
nomatch;
{error, Reason0} ->
log_match_rule_error(Type, Row, Reason0),
nomatch
end
catch
throw:Reason1 ->
log_match_rule_error(Type, Row, Reason1),
nomatch
end.
log_match_rule_error(Type, Row, Reason0) ->
Msg0 = #{
msg => "match_rule_error",
rule => Row,
type => Type
},
Msg1 =
case is_map(Reason0) of
true -> maps:merge(Msg0, Reason0);
false -> Msg0#{reason => Reason0}
end,
?SLOG(
error,
Msg1,
#{tag => "AUTHZ"}
).

8
apps/emqx_auth/test/emqx_authn/emqx_authn_schema_SUITE.erl
View File

@ -122,14 +122,6 @@ t_union_member_selector(_) ->
},
check(BadMechanism)
),
BadCombination = Base#{<<"mechanism">> => <<"scram">>, <<"backend">> => <<"http">>},
?assertThrow(
#{
reason := "unknown_mechanism",
expected := "password_based"
},
check(BadCombination)
),
ok.
t_http_auth_selector(_) ->

5
apps/emqx_auth_http/include/emqx_auth_http.hrl
View File

@ -22,8 +22,13 @@
-define(AUTHN_MECHANISM, password_based).
-define(AUTHN_MECHANISM_BIN, <<"password_based">>).
-define(AUTHN_MECHANISM_SCRAM, scram).
-define(AUTHN_MECHANISM_SCRAM_BIN, <<"scram">>).
-define(AUTHN_BACKEND, http).
-define(AUTHN_BACKEND_BIN, <<"http">>).
-define(AUTHN_TYPE, {?AUTHN_MECHANISM, ?AUTHN_BACKEND}).
-define(AUTHN_TYPE_SCRAM, {?AUTHN_MECHANISM_SCRAM, ?AUTHN_BACKEND}).
-endif.

2
apps/emqx_auth_http/src/emqx_auth_http_app.erl
View File

@ -25,10 +25,12 @@
start(_StartType, _StartArgs) ->
ok = emqx_authz:register_source(?AUTHZ_TYPE, emqx_authz_http),
ok = emqx_authn:register_provider(?AUTHN_TYPE, emqx_authn_http),
ok = emqx_authn:register_provider(?AUTHN_TYPE_SCRAM, emqx_authn_scram_http),
{ok, Sup} = emqx_auth_http_sup:start_link(),
{ok, Sup}.
stop(_State) ->
ok = emqx_authn:deregister_provider(?AUTHN_TYPE),
ok = emqx_authn:deregister_provider(?AUTHN_TYPE_SCRAM),
ok = emqx_authz:unregister_source(?AUTHZ_TYPE),
ok.

7
apps/emqx_auth_http/src/emqx_authn_http.erl
View File

@ -28,6 +28,13 @@
destroy/1
]).
-export([
with_validated_config/2,
generate_request/2,
request_for_log/2,
response_for_log/1
]).
-define(DEFAULT_CONTENT_TYPE, <<"application/json">>).
%%------------------------------------------------------------------------------

8
apps/emqx_auth_http/src/emqx_authn_http_schema.erl
View File

@ -27,6 +27,8 @@
namespace/0
]).
-export([url/1, headers/1, headers_no_content_type/1, request_timeout/1]).
-include("emqx_auth_http.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("hocon/include/hoconsc.hrl").
@ -61,12 +63,6 @@ select_union_member(
got => Else
})
end;
select_union_member(#{<<"backend">> := ?AUTHN_BACKEND_BIN}) ->
throw(#{
reason => "unknown_mechanism",
expected => "password_based",
got => undefined
});
select_union_member(_Value) ->
undefined.

179
apps/emqx_auth_http/src/emqx_authn_scram_http.erl Normal file
View File

@ -0,0 +1,179 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_http).
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("emqx/include/logger.hrl").
-behaviour(emqx_authn_provider).
-export([
create/2,
update/2,
authenticate/2,
destroy/1
]).
-define(REQUIRED_USER_INFO_KEYS, [
<<"stored_key">>,
<<"server_key">>,
<<"salt">>
]).
-define(OPTIONAL_USER_INFO_KEYS, [
<<"is_superuser">>
]).
%%------------------------------------------------------------------------------
%% APIs
%%------------------------------------------------------------------------------
create(_AuthenticatorID, Config) ->
create(Config).
create(Config0) ->
emqx_authn_http:with_validated_config(Config0, fun(Config, State) ->
ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
% {Config, State} = parse_config(Config0),
{ok, _Data} = emqx_authn_utils:create_resource(
ResourceId,
emqx_bridge_http_connector,
Config
),
{ok, merge_scram_conf(Config, State#{resource_id => ResourceId})}
end).
update(Config0, #{resource_id := ResourceId} = _State) ->
emqx_authn_http:with_validated_config(Config0, fun(Config, NState) ->
% {Config, NState} = parse_config(Config0),
case emqx_authn_utils:update_resource(emqx_bridge_http_connector, Config, ResourceId) of
{error, Reason} ->
error({load_config_error, Reason});
{ok, _} ->
{ok, merge_scram_conf(Config, NState#{resource_id => ResourceId})}
end
end).
authenticate(
#{
auth_method := AuthMethod,
auth_data := AuthData,
auth_cache := AuthCache
} = Credential,
State
) ->
RetrieveFun = fun(Username) ->
retrieve(Username, Credential, State)
end,
OnErrFun = fun(Msg, Reason) ->
?TRACE_AUTHN_PROVIDER(Msg, #{
reason => Reason
})
end,
emqx_utils_scram:authenticate(AuthMethod, AuthData, AuthCache, RetrieveFun, OnErrFun, State);
authenticate(_Credential, _State) ->
ignore.
destroy(#{resource_id := ResourceId}) ->
_ = emqx_resource:remove_local(ResourceId),
ok.
%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
retrieve(
Username,
Credential,
#{
resource_id := ResourceId,
method := Method,
request_timeout := RequestTimeout
} = State
) ->
Request = emqx_authn_http:generate_request(Credential#{username := Username}, State),
Response = emqx_resource:simple_sync_query(ResourceId, {Method, Request, RequestTimeout}),
?TRACE_AUTHN_PROVIDER("scram_http_response", #{
request => emqx_authn_http:request_for_log(Credential, State),
response => emqx_authn_http:response_for_log(Response),
resource => ResourceId
}),
case Response of
{ok, 200, Headers, Body} ->
handle_response(Headers, Body);
{ok, _StatusCode, _Headers} ->
{error, bad_response};
{ok, _StatusCode, _Headers, _Body} ->
{error, bad_response};
{error, _Reason} = Error ->
Error
end.
handle_response(Headers, Body) ->
ContentType = proplists:get_value(<<"content-type">>, Headers),
case safely_parse_body(ContentType, Body) of
{ok, NBody} ->
body_to_user_info(NBody);
{error, Reason} = Error ->
?TRACE_AUTHN_PROVIDER(
error,
"parse_scram_http_response_failed",
#{content_type => ContentType, body => Body, reason => Reason}
),
Error
end.
body_to_user_info(Body) ->
Required0 = maps:with(?REQUIRED_USER_INFO_KEYS, Body),
case maps:size(Required0) =:= erlang:length(?REQUIRED_USER_INFO_KEYS) of
true ->
case safely_convert_hex(Required0) of
{ok, Required} ->
UserInfo0 = maps:merge(Required, maps:with(?OPTIONAL_USER_INFO_KEYS, Body)),
UserInfo1 = emqx_utils_maps:safe_atom_key_map(UserInfo0),
UserInfo = maps:merge(#{is_superuser => false}, UserInfo1),
{ok, UserInfo};
Error ->
Error
end;
_ ->
?TRACE_AUTHN_PROVIDER("bad_response_body", #{http_body => Body}),
{error, bad_response}
end.
safely_parse_body(ContentType, Body) ->
try
parse_body(ContentType, Body)
catch
_Class:_Reason ->
{error, invalid_body}
end.
safely_convert_hex(Required) ->
try
{ok,
maps:map(
fun(_Key, Hex) ->
binary:decode_hex(Hex)
end,
Required
)}
catch
_Class:Reason ->
{error, Reason}
end.
parse_body(<<"application/json", _/binary>>, Body) ->
{ok, emqx_utils_json:decode(Body, [return_maps])};
parse_body(<<"application/x-www-form-urlencoded", _/binary>>, Body) ->
Flags = ?REQUIRED_USER_INFO_KEYS ++ ?OPTIONAL_USER_INFO_KEYS,
RawMap = maps:from_list(cow_qs:parse_qs(Body)),
NBody = maps:with(Flags, RawMap),
{ok, NBody};
parse_body(ContentType, _) ->
{error, {unsupported_content_type, ContentType}}.
merge_scram_conf(Conf, State) ->
maps:merge(maps:with([algorithm, iteration_count], Conf), State).

81
apps/emqx_auth_http/src/emqx_authn_scram_http_schema.erl Normal file
View File

@ -0,0 +1,81 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_http_schema).
-behaviour(emqx_authn_schema).
-export([
fields/1,
validations/0,
desc/1,
refs/0,
select_union_member/1,
namespace/0
]).
-include("emqx_auth_http.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("hocon/include/hoconsc.hrl").
namespace() -> "authn".
refs() ->
[?R_REF(scram_http_get), ?R_REF(scram_http_post)].
select_union_member(
#{<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN, <<"backend">> := ?AUTHN_BACKEND_BIN} = Value
) ->
case maps:get(<<"method">>, Value, undefined) of
<<"get">> ->
[?R_REF(scram_http_get)];
<<"post">> ->
[?R_REF(scramm_http_post)];
Else ->
throw(#{
reason => "unknown_http_method",
expected => "get | post",
field_name => method,
got => Else
})
end;
select_union_member(_Value) ->
undefined.
fields(scram_http_get) ->
[
{method, #{type => get, required => true, desc => ?DESC(emqx_authn_http_schema, method)}},
{headers, fun emqx_authn_http_schema:headers_no_content_type/1}
] ++ common_fields();
fields(scram_http_post) ->
[
{method, #{type => post, required => true, desc => ?DESC(emqx_authn_http_schema, method)}},
{headers, fun emqx_authn_http_schema:headers/1}
] ++ common_fields().
desc(scram_http_get) ->
?DESC(emqx_authn_http_schema, get);
desc(scram_http_post) ->
?DESC(emqx_authn_http_schema, post);
desc(_) ->
undefined.
validations() ->
emqx_authn_http_schema:validations().
common_fields() ->
emqx_authn_schema:common_fields() ++
[
{mechanism, emqx_authn_schema:mechanism(?AUTHN_MECHANISM_SCRAM)},
{backend, emqx_authn_schema:backend(?AUTHN_BACKEND)},
{algorithm, fun emqx_authn_scram_mnesia_schema:algorithm/1},
{iteration_count, fun emqx_authn_scram_mnesia_schema:iteration_count/1},
{url, fun emqx_authn_http_schema:url/1},
{body,
hoconsc:mk(typerefl:alias("map", map([{fuzzy, term(), binary()}])), #{
required => false, desc => ?DESC(emqx_authn_http_schema, body)
})},
{request_timeout, fun emqx_authn_http_schema:request_timeout/1}
] ++
proplists:delete(pool_type, emqx_bridge_http_connector:fields(config)).

6
apps/emqx_auth_http/src/emqx_authz_http.erl
View File

@ -67,7 +67,11 @@ description() ->
create(Config) ->
NConfig = parse_config(Config),
ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
{ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_bridge_http_connector, NConfig),
{ok, _Data} = emqx_authz_utils:create_resource(
ResourceId,
emqx_bridge_http_connector,
NConfig
),
NConfig#{annotations => #{id => ResourceId}}.
update(Config) ->

438
apps/emqx_auth_http/test/emqx_authn_scram_http_SUITE.erl Normal file
View File

@ -0,0 +1,438 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_http_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("eunit/include/eunit.hrl").
-include_lib("common_test/include/ct.hrl").
-include_lib("emqx/include/emqx_mqtt.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-define(PATH, [authentication]).
-define(HTTP_PORT, 34333).
-define(HTTP_PATH, "/user/[...]").
-define(ALGORITHM, sha512).
-define(ALGORITHM_STR, <<"sha512">>).
-define(ITERATION_COUNT, 4096).
-include_lib("emqx/include/emqx_placeholder.hrl").
all() ->
case emqx_release:edition() of
ce ->
[];
_ ->
emqx_common_test_helpers:all(?MODULE)
end.
init_per_suite(Config) ->
Apps = emqx_cth_suite:start([cowboy, emqx, emqx_conf, emqx_auth, emqx_auth_http], #{
work_dir => ?config(priv_dir, Config)
}),
IdleTimeout = emqx_config:get([mqtt, idle_timeout]),
[{apps, Apps}, {idle_timeout, IdleTimeout} | Config].
end_per_suite(Config) ->
ok = emqx_config:put([mqtt, idle_timeout], ?config(idle_timeout, Config)),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
ok = emqx_cth_suite:stop(?config(apps, Config)),
ok.
init_per_testcase(_Case, Config) ->
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
{ok, _} = emqx_authn_scram_http_test_server:start_link(?HTTP_PORT, ?HTTP_PATH),
Config.
end_per_testcase(_Case, _Config) ->
ok = emqx_authn_scram_http_test_server:stop().
%%------------------------------------------------------------------------------
%% Tests
%%------------------------------------------------------------------------------
t_create(_Config) ->
AuthConfig = raw_config(),
{ok, _} = emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, AuthConfig}
),
{ok, [#{provider := emqx_authn_scram_http}]} = emqx_authn_chains:list_authenticators(?GLOBAL).
t_create_invalid(_Config) ->
AuthConfig = raw_config(),
InvalidConfigs =
[
AuthConfig#{<<"headers">> => []},
AuthConfig#{<<"method">> => <<"delete">>},
AuthConfig#{<<"url">> => <<"localhost">>},
AuthConfig#{<<"url">> => <<"http://foo.com/xxx#fragment">>},
AuthConfig#{<<"url">> => <<"http://${foo}.com/xxx">>},
AuthConfig#{<<"url">> => <<"//foo.com/xxx">>},
AuthConfig#{<<"algorithm">> => <<"sha128">>}
],
lists:foreach(
fun(Config) ->
ct:pal("creating authenticator with invalid config: ~p", [Config]),
{error, _} =
try
emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, Config}
)
catch
throw:Error ->
{error, Error}
end,
?assertEqual(
{error, {not_found, {chain, ?GLOBAL}}},
emqx_authn_chains:list_authenticators(?GLOBAL)
)
end,
InvalidConfigs
).
t_authenticate(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
ok = emqx_config:put([mqtt, idle_timeout], 500),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(Username),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
%% Intentional sleep to trigger idle timeout for the connection not yet authenticated
ok = ct:sleep(1000),
?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{'Authentication-Data' := ServerFirstMessage}
) = receive_packet(),
{continue, ClientFinalMessage, ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => Password,
algorithm => ?ALGORITHM
}
),
AuthContinuePacket = ?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFinalMessage
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
?CONNACK_PACKET(
?RC_SUCCESS,
_,
#{'Authentication-Data' := ServerFinalMessage}
) = receive_packet(),
ok = esasl_scram:check_server_final_message(
ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
).
t_authenticate_bad_props(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_authenticate_bad_username(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(<<"badusername">>),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_authenticate_bad_password(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(Username),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{'Authentication-Data' := ServerFirstMessage}
) = receive_packet(),
{continue, ClientFinalMessage, _ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => <<"badpassword">>,
algorithm => ?ALGORITHM
}
),
AuthContinuePacket = ?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFinalMessage
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_destroy(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
ok = emqx_config:put([mqtt, idle_timeout], 500),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
ok = ct:sleep(1000),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet(),
%% emqx_authn_mqtt_test_client:stop(Pid),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
{ok, Pid2} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ok = emqx_authn_mqtt_test_client:send(Pid2, ConnectPacket),
ok = ct:sleep(1000),
?CONNACK_PACKET(
?RC_SUCCESS,
_,
_
) = receive_packet().
t_is_superuser() ->
State = init_auth(),
ok = test_is_superuser(State, false),
ok = test_is_superuser(State, true),
ok = test_is_superuser(State, false).
test_is_superuser(State, ExpectedIsSuperuser) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password, ExpectedIsSuperuser),
ClientFirstMessage = esasl_scram:client_first_message(Username),
{continue, ServerFirstMessage, ServerCache} =
emqx_authn_scram_http:authenticate(
#{
auth_method => <<"SCRAM-SHA-512">>,
auth_data => ClientFirstMessage,
auth_cache => #{}
},
State
),
{continue, ClientFinalMessage, ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => Password,
algorithm => ?ALGORITHM
}
),
{ok, UserInfo1, ServerFinalMessage} =
emqx_authn_scram_http:authenticate(
#{
auth_method => <<"SCRAM-SHA-512">>,
auth_data => ClientFinalMessage,
auth_cache => ServerCache
},
State
),
ok = esasl_scram:check_server_final_message(
ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
),
?assertMatch(#{is_superuser := ExpectedIsSuperuser}, UserInfo1).
%%------------------------------------------------------------------------------
%% Helpers
%%------------------------------------------------------------------------------
raw_config() ->
#{
<<"mechanism">> => <<"scram">>,
<<"backend">> => <<"http">>,
<<"enable">> => <<"true">>,
<<"method">> => <<"get">>,
<<"url">> => <<"http://127.0.0.1:34333/user">>,
<<"body">> => #{<<"username">> => ?PH_USERNAME},
<<"headers">> => #{<<"X-Test-Header">> => <<"Test Value">>},
<<"algorithm">> => ?ALGORITHM_STR,
<<"iteration_count">> => ?ITERATION_COUNT
}.
set_user_handler(Username, Password) ->
set_user_handler(Username, Password, false).
set_user_handler(Username, Password, IsSuperuser) ->
%% HTTP Server
Handler = fun(Req0, State) ->
#{
username := Username
} = cowboy_req:match_qs([username], Req0),
UserInfo = make_user_info(Password, ?ALGORITHM, ?ITERATION_COUNT, IsSuperuser),
Req = cowboy_req:reply(
200,
#{<<"content-type">> => <<"application/json">>},
emqx_utils_json:encode(UserInfo),
Req0
),
{ok, Req, State}
end,
ok = emqx_authn_scram_http_test_server:set_handler(Handler).
init_auth() ->
init_auth(raw_config()).
init_auth(Config) ->
{ok, _} = emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, Config}
),
{ok, [#{state := State}]} = emqx_authn_chains:list_authenticators(?GLOBAL),
State.
make_user_info(Password, Algorithm, IterationCount, IsSuperuser) ->
{StoredKey, ServerKey, Salt} = esasl_scram:generate_authentication_info(
Password,
#{
algorithm => Algorithm,
iteration_count => IterationCount
}
),
#{
stored_key => binary:encode_hex(StoredKey),
server_key => binary:encode_hex(ServerKey),
salt => binary:encode_hex(Salt),
is_superuser => IsSuperuser
}.
receive_packet() ->
receive
{packet, Packet} ->
ct:pal("Delivered packet: ~p", [Packet]),
Packet
after 1000 ->
ct:fail("Deliver timeout")
end.

115
apps/emqx_auth_http/test/emqx_authn_scram_http_test_server.erl Normal file
View File

@ -0,0 +1,115 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_http_test_server).
-behaviour(supervisor).
-behaviour(cowboy_handler).
% cowboy_server callbacks
-export([init/2]).
% supervisor callbacks
-export([init/1]).
% API
-export([
start_link/2,
start_link/3,
stop/0,
set_handler/1
]).
%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------
start_link(Port, Path) ->
start_link(Port, Path, false).
start_link(Port, Path, SSLOpts) ->
supervisor:start_link({local, ?MODULE}, ?MODULE, [Port, Path, SSLOpts]).
stop() ->
gen_server:stop(?MODULE).
set_handler(F) when is_function(F, 2) ->
true = ets:insert(?MODULE, {handler, F}),
ok.
%%------------------------------------------------------------------------------
%% supervisor API
%%------------------------------------------------------------------------------
init([Port, Path, SSLOpts]) ->
Dispatch = cowboy_router:compile(
[
{'_', [{Path, ?MODULE, []}]}
]
),
ProtoOpts = #{env => #{dispatch => Dispatch}},
Tab = ets:new(?MODULE, [set, named_table, public]),
ets:insert(Tab, {handler, fun default_handler/2}),
{Transport, TransOpts, CowboyModule} = transport_settings(Port, SSLOpts),
ChildSpec = ranch:child_spec(?MODULE, Transport, TransOpts, CowboyModule, ProtoOpts),
{ok, {#{}, [ChildSpec]}}.
%%------------------------------------------------------------------------------
%% cowboy_server API
%%------------------------------------------------------------------------------
init(Req, State) ->
[{handler, Handler}] = ets:lookup(?MODULE, handler),
Handler(Req, State).
%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------
transport_settings(Port, false) ->
TransOpts = #{
socket_opts => [{port, Port}],
connection_type => supervisor
},
{ranch_tcp, TransOpts, cowboy_clear};
transport_settings(Port, SSLOpts) ->
TransOpts = #{
socket_opts => [
{port, Port},
{next_protocols_advertised, [<<"h2">>, <<"http/1.1">>]},
{alpn_preferred_protocols, [<<"h2">>, <<"http/1.1">>]}
| SSLOpts
],
connection_type => supervisor
},
{ranch_ssl, TransOpts, cowboy_tls}.
default_handler(Req0, State) ->
Req = cowboy_req:reply(
400,
#{<<"content-type">> => <<"text/plain">>},
<<"">>,
Req0
),
{ok, Req, State}.
make_user_info(Password, Algorithm, IterationCount) ->
{StoredKey, ServerKey, Salt} = esasl_scram:generate_authentication_info(
Password,
#{
algorithm => Algorithm,
iteration_count => IterationCount
}
),
#{
stored_key => StoredKey,
server_key => ServerKey,
salt => Salt,
is_superuser => false
}.

3
apps/emqx_auth_jwt/src/emqx_authn_jwks_connector.erl
View File

@ -22,6 +22,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -32,6 +33,8 @@
-define(DEFAULT_POOL_SIZE, 8).
resource_type() -> jwks.
callback_mode() -> always_sync.
on_start(InstId, Opts) ->

3
apps/emqx_auth_jwt/src/emqx_authn_jwt.erl
View File

@ -188,7 +188,8 @@ do_create(
ResourceId,
?AUTHN_RESOURCE_GROUP,
emqx_authn_jwks_connector,
connector_opts(Config)
connector_opts(Config),
#{}
),
{ok, #{
jwk_resource => ResourceId,

2
apps/emqx_auth_ldap/src/emqx_auth_ldap.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_auth_ldap, [
{description, "EMQX LDAP Authentication and Authorization"},
{vsn, "0.1.2"},
{vsn, "0.1.3"},
{registered, []},
{mod, {emqx_auth_ldap_app, []}},
{applications, [

69
apps/emqx_auth_mnesia/src/emqx_authn_scram_mnesia.erl
View File

@ -133,17 +133,15 @@ authenticate(
},
State
) ->
case ensure_auth_method(AuthMethod, AuthData, State) of
true ->
case AuthCache of
#{next_step := client_final} ->
check_client_final_message(AuthData, AuthCache, State);
_ ->
check_client_first_message(AuthData, AuthCache, State)
end;
false ->
ignore
end;
RetrieveFun = fun(Username) ->
retrieve(Username, State)
end,
OnErrFun = fun(Msg, Reason) ->
?TRACE_AUTHN_PROVIDER(Msg, #{
reason => Reason
})
end,
emqx_utils_scram:authenticate(AuthMethod, AuthData, AuthCache, RetrieveFun, OnErrFun, State);
authenticate(_Credential, _State) ->
ignore.
@ -257,55 +255,6 @@ run_fuzzy_filter(
%% Internal functions
%%------------------------------------------------------------------------------
ensure_auth_method(_AuthMethod, undefined, _State) ->
false;
ensure_auth_method(<<"SCRAM-SHA-256">>, _AuthData, #{algorithm := sha256}) ->
true;
ensure_auth_method(<<"SCRAM-SHA-512">>, _AuthData, #{algorithm := sha512}) ->
true;
ensure_auth_method(_AuthMethod, _AuthData, _State) ->
false.
check_client_first_message(Bin, _Cache, #{iteration_count := IterationCount} = State) ->
RetrieveFun = fun(Username) ->
retrieve(Username, State)
end,
case
esasl_scram:check_client_first_message(
Bin,
#{
iteration_count => IterationCount,
retrieve => RetrieveFun
}
)
of
{continue, ServerFirstMessage, Cache} ->
{continue, ServerFirstMessage, Cache};
ignore ->
ignore;
{error, Reason} ->
?TRACE_AUTHN_PROVIDER("check_client_first_message_error", #{
reason => Reason
}),
{error, not_authorized}
end.
check_client_final_message(Bin, #{is_superuser := IsSuperuser} = Cache, #{algorithm := Alg}) ->
case
esasl_scram:check_client_final_message(
Bin,
Cache#{algorithm => Alg}
)
of
{ok, ServerFinalMessage} ->
{ok, #{is_superuser => IsSuperuser}, ServerFinalMessage};
{error, Reason} ->
?TRACE_AUTHN_PROVIDER("check_client_final_message_error", #{
reason => Reason
}),
{error, not_authorized}
end.
user_info_record(
#{
user_id := UserID,

7
apps/emqx_auth_mnesia/src/emqx_authn_scram_mnesia_schema.erl
View File

@ -29,6 +29,8 @@
select_union_member/1
]).
-export([algorithm/1, iteration_count/1]).
namespace() -> "authn".
refs() ->
@ -38,11 +40,6 @@ select_union_member(#{
<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN, <<"backend">> := ?AUTHN_BACKEND_BIN
}) ->
refs();
select_union_member(#{<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN}) ->
throw(#{
reason => "unknown_backend",
expected => ?AUTHN_BACKEND
});
select_union_member(_) ->
undefined.

20
apps/emqx_auth_mysql/src/emqx_authz_mysql.erl
View File

@ -101,19 +101,9 @@ authorize(
do_authorize(_Client, _Action, _Topic, _ColumnNames, []) ->
nomatch;
do_authorize(Client, Action, Topic, ColumnNames, [Row | Tail]) ->
try
emqx_authz_rule:match(
Client, Action, Topic, emqx_authz_utils:parse_rule_from_row(ColumnNames, Row)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, ColumnNames, Tail)
catch
error:Reason ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule => Row
}),
do_authorize(Client, Action, Topic, ColumnNames, Tail)
case emqx_authz_utils:do_authorize(mysql, Client, Action, Topic, ColumnNames, Row) of
nomatch ->
do_authorize(Client, Action, Topic, ColumnNames, Tail);
{matched, Permission} ->
{matched, Permission}
end.

21
apps/emqx_auth_postgresql/src/emqx_authz_postgresql.erl
View File

@ -107,22 +107,11 @@ authorize(
do_authorize(_Client, _Action, _Topic, _ColumnNames, []) ->
nomatch;
do_authorize(Client, Action, Topic, ColumnNames, [Row | Tail]) ->
try
emqx_authz_rule:match(
Client, Action, Topic, emqx_authz_utils:parse_rule_from_row(ColumnNames, Row)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, ColumnNames, Tail)
catch
error:Reason:Stack ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule => Row,
stack => Stack
}),
do_authorize(Client, Action, Topic, ColumnNames, Tail)
case emqx_authz_utils:do_authorize(postgresql, Client, Action, Topic, ColumnNames, Row) of
nomatch ->
do_authorize(Client, Action, Topic, ColumnNames, Tail);
{matched, Permission} ->
{matched, Permission}
end.
column_names(Columns) ->

4
apps/emqx_auth_postgresql/test/emqx_authn_postgresql_SUITE.erl
View File

@ -198,9 +198,9 @@ test_user_auth(#{
t_authenticate_disabled_prepared_statements(_Config) ->
ResConfig = maps:merge(pgsql_config(), #{disable_prepared_statements => true}),
{ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig),
{ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig, #{}),
on_exit(fun() ->
emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config())
emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config(), #{})
end),
ok = lists:foreach(
fun(Sample0) ->

68
apps/emqx_auth_redis/src/emqx_authz_redis.erl
View File

@ -92,44 +92,30 @@ authorize(
do_authorize(_Client, _Action, _Topic, []) ->
nomatch;
do_authorize(Client, Action, Topic, [TopicFilterRaw, RuleEncoded | Tail]) ->
try
emqx_authz_rule:match(
Client,
Action,
Topic,
compile_rule(RuleEncoded, TopicFilterRaw)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, Tail)
catch
error:Reason:Stack ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule_encoded => RuleEncoded,
topic_filter_raw => TopicFilterRaw,
stacktrace => Stack
case parse_rule(RuleEncoded) of
{ok, RuleMap0} ->
RuleMap =
maps:merge(
#{
<<"permission">> => <<"allow">>,
<<"topic">> => TopicFilterRaw
},
RuleMap0
),
case emqx_authz_utils:do_authorize(redis, Client, Action, Topic, undefined, RuleMap) of
nomatch ->
do_authorize(Client, Action, Topic, Tail);
{matched, Permission} ->
{matched, Permission}
end;
{error, Reason} ->
?SLOG(error, Reason#{
msg => "parse_rule_error",
rule => RuleEncoded
}),
do_authorize(Client, Action, Topic, Tail)
end.
compile_rule(RuleBin, TopicFilterRaw) ->
RuleRaw =
maps:merge(
#{
<<"permission">> => <<"allow">>,
<<"topic">> => TopicFilterRaw
},
parse_rule(RuleBin)
),
case emqx_authz_rule_raw:parse_rule(RuleRaw) of
{ok, {Permission, Action, Topics}} ->
emqx_authz_rule:compile({Permission, all, Action, Topics});
{error, Reason} ->
error(Reason)
end.
parse_cmd(Query) ->
case emqx_redis_command:split(Query) of
{ok, Cmd} ->
@ -154,17 +140,17 @@ validate_cmd(Cmd) ->
end.
parse_rule(<<"publish">>) ->
#{<<"action">> => <<"publish">>};
{ok, #{<<"action">> => <<"publish">>}};
parse_rule(<<"subscribe">>) ->
#{<<"action">> => <<"subscribe">>};
{ok, #{<<"action">> => <<"subscribe">>}};
parse_rule(<<"all">>) ->
#{<<"action">> => <<"all">>};
{ok, #{<<"action">> => <<"all">>}};
parse_rule(Bin) when is_binary(Bin) ->
case emqx_utils_json:safe_decode(Bin, [return_maps]) of
{ok, Map} when is_map(Map) ->
maps:with([<<"qos">>, <<"action">>, <<"retain">>], Map);
{ok, maps:with([<<"qos">>, <<"action">>, <<"retain">>], Map)};
{ok, _} ->
error({invalid_topic_rule, Bin, notamap});
{error, Error} ->
error({invalid_topic_rule, Bin, Error})
{error, #{reason => invalid_topic_rule_not_map, value => Bin}};
{error, _Error} ->
{error, #{reason => invalid_topic_rule_not_json, value => Bin}}
end.

4
apps/emqx_bridge/src/emqx_bridge_resource.erl
View File

@ -196,7 +196,7 @@ create(Type, Name, Conf0, Opts) ->
Conf = Conf0#{bridge_type => TypeBin, bridge_name => Name},
{ok, _Data} = emqx_resource:create_local(
resource_id(Type, Name),
<<"emqx_bridge">>,
<<"bridge">>,
bridge_to_resource_type(Type),
parse_confs(TypeBin, Name, Conf),
parse_opts(Conf, Opts)
@ -282,7 +282,7 @@ create_dry_run(Type0, Conf0) ->
create_dry_run_bridge_v1(Type, Conf0) ->
TmpName = iolist_to_binary([?TEST_ID_PREFIX, emqx_utils:gen_id(8)]),
TmpPath = emqx_utils:safe_filename(TmpName),
%% Already typechecked, no need to catch errors
%% Already type checked, no need to catch errors
TypeBin = bin(Type),
TypeAtom = safe_atom(Type),
Conf1 = maps:without([<<"name">>], Conf0),

1
apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl
View File

@ -1110,6 +1110,7 @@ t_query_uses_action_query_mode(_Config) ->
%% ... now we use a quite different query mode for the action
meck:expect(con_mod(), query_mode, 1, simple_async_internal_buffer),
meck:expect(con_mod(), resource_type, 0, dummy),
meck:expect(con_mod(), callback_mode, 0, async_if_possible),
{ok, _} = emqx_bridge_v2:create(bridge_type(), ActionName, ActionConfig),

1
apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl
View File

@ -293,6 +293,7 @@ init_mocks() ->
meck:new(emqx_connector_resource, [passthrough, no_link]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
meck:expect(?CONNECTOR_IMPL, resource_type, 0, dummy),
meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
meck:expect(
?CONNECTOR_IMPL,

4
apps/emqx_bridge/test/emqx_bridge_v2_dummy_connector.erl
View File

@ -15,15 +15,17 @@
%% this module is only intended to be mocked
-module(emqx_bridge_v2_dummy_connector).
-behavior(emqx_resource).
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
on_add_channel/4,
on_get_channel_status/3
]).
resource_type() -> dummy.
callback_mode() -> error(unexpected).
on_start(_, _) -> error(unexpected).
on_stop(_, _) -> error(unexpected).

3
apps/emqx_bridge/test/emqx_bridge_v2_test_connector.erl
View File

@ -19,6 +19,7 @@
-export([
query_mode/1,
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -34,6 +35,8 @@
query_mode(_Config) ->
sync.
resource_type() -> test_connector.
callback_mode() ->
always_sync.

2
apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_cassandra, [
{description, "EMQX Enterprise Cassandra Bridge"},
{vsn, "0.3.1"},
{vsn, "0.3.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra_connector.erl
View File

@ -19,6 +19,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -94,6 +95,7 @@ desc("connector") ->
%%--------------------------------------------------------------------
%% callbacks for emqx_resource
resource_type() -> cassandra.
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_clickhouse, [
{description, "EMQX Enterprise ClickHouse Bridge"},
{vsn, "0.4.1"},
{vsn, "0.4.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse_connector.erl
View File

@ -29,6 +29,7 @@
%% callbacks for behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -128,6 +129,7 @@ values(_) ->
%% ===================================================================
%% Callbacks defined in emqx_resource
%% ===================================================================
resource_type() -> clickhouse.
callback_mode() -> always_sync.

2
apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_dynamo, [
{description, "EMQX Enterprise Dynamo Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo_connector.erl
View File

@ -17,6 +17,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -68,6 +69,7 @@ fields(config) ->
%%========================================================================================
%% `emqx_resource' API
%%========================================================================================
resource_type() -> dynamo.
callback_mode() -> always_sync.

2
apps/emqx_bridge_es/src/emqx_bridge_es.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_bridge_es, [
{description, "EMQX Enterprise Elastic Search Bridge"},
{vsn, "0.1.3"},
{vsn, "0.1.4"},
{modules, [
emqx_bridge_es,
emqx_bridge_es_connector

3
apps/emqx_bridge_es/src/emqx_bridge_es_connector.erl
View File

@ -14,6 +14,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -207,6 +208,8 @@ base_url(#{server := Server}) -> "http://" ++ Server.
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> elastic_search.
callback_mode() -> async_if_possible.
-spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

3
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_consumer.erl
View File

@ -8,6 +8,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -84,6 +85,8 @@
%%-------------------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------------------
-spec resource_type() -> resource_type().
resource_type() -> gcp_pubsub_consumer.
-spec callback_mode() -> callback_mode().
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_producer.erl
View File

@ -41,6 +41,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -62,6 +63,7 @@
%%-------------------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------------------
resource_type() -> gcp_pubsub.
callback_mode() -> async_if_possible.

3
apps/emqx_bridge_greptimedb/src/emqx_bridge_greptimedb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -67,6 +68,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> greptimedb.
callback_mode() -> async_if_possible.
on_add_channel(

2
apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_hstreamdb, [
{description, "EMQX Enterprise HStreamDB Bridge"},
{vsn, "0.2.1"},
{vsn, "0.2.2"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -44,6 +45,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> hstreamdb.
callback_mode() -> always_sync.
on_start(InstId, Config) ->

2
apps/emqx_bridge_http/src/emqx_bridge_http_connector.erl
View File

@ -26,6 +26,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -183,6 +184,7 @@ sc(Type, Meta) -> hoconsc:mk(Type, Meta).
ref(Field) -> hoconsc:ref(?MODULE, Field).
%% ===================================================================
resource_type() -> webhook.
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_influxdb, [
{description, "EMQX Enterprise InfluxDB Bridge"},
{vsn, "0.2.3"},
{vsn, "0.2.4"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -70,6 +71,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> influxdb.
callback_mode() -> async_if_possible.
on_add_channel(

2
apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_bridge_iotdb, [
{description, "EMQX Enterprise Apache IoTDB Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{modules, [
emqx_bridge_iotdb,
emqx_bridge_iotdb_connector

3
apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_connector.erl
View File

@ -15,6 +15,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -206,6 +207,8 @@ proplists_without(Keys, List) ->
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> iotdb.
callback_mode() -> async_if_possible.
-spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

14
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_consumer.erl
View File

@ -7,6 +7,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -126,6 +127,7 @@
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> kafka_consumer.
callback_mode() ->
async_if_possible.
@ -631,16 +633,6 @@ consumer_group_id(_ConsumerParams, BridgeName0) ->
BridgeName = to_bin(BridgeName0),
<<"emqx-kafka-consumer-", BridgeName/binary>>.
-spec is_dry_run(connector_resource_id()) -> boolean().
is_dry_run(ConnectorResId) ->
TestIdStart = string:find(ConnectorResId, ?TEST_ID_PREFIX),
case TestIdStart of
nomatch ->
false;
_ ->
string:equal(TestIdStart, ConnectorResId)
end.
-spec check_client_connectivity(pid()) ->
?status_connected
| ?status_disconnected
@ -676,7 +668,7 @@ maybe_clean_error(Reason) ->
-spec make_client_id(connector_resource_id(), binary(), atom() | binary()) -> atom().
make_client_id(ConnectorResId, BridgeType, BridgeName) ->
case is_dry_run(ConnectorResId) of
case emqx_resource:is_dry_run(ConnectorResId) of
false ->
ClientID0 = emqx_bridge_kafka_impl:make_client_id(BridgeType, BridgeName),
binary_to_atom(ClientID0);

12
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
View File

@ -10,6 +10,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
query_mode/1,
callback_mode/0,
on_start/2,
@ -35,6 +36,8 @@
-define(kafka_client_id, kafka_client_id).
-define(kafka_producers, kafka_producers).
resource_type() -> kafka_producer.
query_mode(#{parameters := #{query_mode := sync}}) ->
simple_sync_internal_buffer;
query_mode(_) ->
@ -140,14 +143,7 @@ create_producers_for_bridge_v2(
KafkaHeadersValEncodeMode = maps:get(kafka_header_value_encode_mode, KafkaConfig, none),
MaxPartitions = maps:get(partitions_limit, KafkaConfig, all_partitions),
#{name := BridgeName} = emqx_bridge_v2:parse_id(BridgeV2Id),
TestIdStart = string:find(BridgeV2Id, ?TEST_ID_PREFIX),
IsDryRun =
case TestIdStart of
nomatch ->
false;
_ ->
string:equal(TestIdStart, InstId)
end,
IsDryRun = emqx_resource:is_dry_run(BridgeV2Id),
ok = check_topic_and_leader_connections(ClientId, KafkaTopic, MaxPartitions),
WolffProducerConfig = producers_config(
BridgeType, BridgeName, KafkaConfig, IsDryRun, BridgeV2Id

2
apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_kinesis, [
{description, "EMQX Enterprise Amazon Kinesis Bridge"},
{vsn, "0.2.1"},
{vsn, "0.2.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis_impl_producer.erl
View File

@ -30,6 +30,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -50,6 +51,7 @@
%%-------------------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------------------
resource_type() -> kinesis_producer.
callback_mode() -> always_sync.

2
apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_mongodb, [
{description, "EMQX Enterprise MongoDB Bridge"},
{vsn, "0.3.2"},
{vsn, "0.3.3"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb_connector.erl
View File

@ -11,6 +11,7 @@
%% `emqx_resource' API
-export([
on_remove_channel/3,
resource_type/0,
callback_mode/0,
on_add_channel/4,
on_get_channel_status/3,
@ -25,6 +26,7 @@
%%========================================================================================
%% `emqx_resource' API
%%========================================================================================
resource_type() -> emqx_mongodb:resource_type().
callback_mode() -> emqx_mongodb:callback_mode().

2
apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_bridge_mqtt, [
{description, "EMQX MQTT Broker Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

20
apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector.erl
View File

@ -30,6 +30,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -76,6 +77,8 @@ on_message_received(Msg, HookPoints, ResId) ->
ok.
%% ===================================================================
resource_type() -> mqtt.
callback_mode() -> async_if_possible.
on_start(ResourceId, #{server := Server} = Conf) ->
@ -207,7 +210,7 @@ start_mqtt_clients(ResourceId, Conf) ->
start_mqtt_clients(ResourceId, Conf, ClientOpts).
start_mqtt_clients(ResourceId, StartConf, ClientOpts) ->
PoolName = <<ResourceId/binary>>,
PoolName = ResourceId,
#{
pool_size := PoolSize
} = StartConf,
@ -227,7 +230,7 @@ start_mqtt_clients(ResourceId, StartConf, ClientOpts) ->
on_stop(ResourceId, State) ->
?SLOG(info, #{
msg => "stopping_mqtt_connector",
connector => ResourceId
resource_id => ResourceId
}),
%% on_stop can be called with State = undefined
StateMap =
@ -271,7 +274,7 @@ on_query(
on_query(ResourceId, {_ChannelId, Msg}, #{}) ->
?SLOG(error, #{
msg => "forwarding_unavailable",
connector => ResourceId,
resource_id => ResourceId,
message => Msg,
reason => "Egress is not configured"
}).
@ -298,7 +301,7 @@ on_query_async(
on_query_async(ResourceId, {_ChannelId, Msg}, _Callback, #{}) ->
?SLOG(error, #{
msg => "forwarding_unavailable",
connector => ResourceId,
resource_id => ResourceId,
message => Msg,
reason => "Egress is not configured"
}).
@ -463,8 +466,10 @@ connect(Options) ->
{ok, Pid} ->
connect(Pid, Name);
{error, Reason} = Error ->
?SLOG(error, #{
IsDryRun = emqx_resource:is_dry_run(Name),
?SLOG(?LOG_LEVEL(IsDryRun), #{
msg => "client_start_failed",
resource_id => Name,
config => emqx_utils:redact(ClientOpts),
reason => Reason
}),
@ -508,10 +513,11 @@ connect(Pid, Name) ->
{ok, _Props} ->
{ok, Pid};
{error, Reason} = Error ->
?SLOG(warning, #{
IsDryRun = emqx_resource:is_dry_run(Name),
?SLOG(?LOG_LEVEL(IsDryRun), #{
msg => "ingress_client_connect_failed",
reason => Reason,
name => Name
resource_id => Name
}),
_ = catch emqtt:stop(Pid),
Error

2
apps/emqx_bridge_mysql/src/emqx_bridge_mysql.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_mysql, [
{description, "EMQX Enterprise MySQL Bridge"},
{vsn, "0.1.7"},
{vsn, "0.1.8"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_mysql/src/emqx_bridge_mysql_connector.erl
View File

@ -10,6 +10,7 @@
%% `emqx_resource' API
-export([
on_remove_channel/3,
resource_type/0,
callback_mode/0,
on_add_channel/4,
on_batch_query/3,
@ -24,6 +25,7 @@
%%========================================================================================
%% `emqx_resource' API
%%========================================================================================
resource_type() -> emqx_mysql:resource_type().
callback_mode() -> emqx_mysql:callback_mode().

2
apps/emqx_bridge_opents/src/emqx_bridge_opents.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_opents, [
{description, "EMQX Enterprise OpenTSDB Bridge"},
{vsn, "0.2.1"},
{vsn, "0.2.2"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_bridge_opents/src/emqx_bridge_opents_connector.erl
View File

@ -18,6 +18,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -114,6 +115,8 @@ connector_example_values() ->
-define(HTTP_CONNECT_TIMEOUT, 1000).
resource_type() -> opents.
callback_mode() -> always_sync.
on_start(

2
apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_pulsar, [
{description, "EMQX Pulsar Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

14
apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar_connector.erl
View File

@ -10,6 +10,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -55,6 +56,7 @@
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> pulsar.
callback_mode() -> async_if_possible.
@ -255,7 +257,7 @@ format_servers(Servers0) ->
-spec make_client_id(resource_id()) -> pulsar_client_id().
make_client_id(InstanceId) ->
case is_dry_run(InstanceId) of
case emqx_resource:is_dry_run(InstanceId) of
true ->
pulsar_producer_probe;
false ->
@ -269,14 +271,6 @@ make_client_id(InstanceId) ->
binary_to_atom(ClientIdBin)
end.
-spec is_dry_run(resource_id()) -> boolean().
is_dry_run(InstanceId) ->
TestIdStart = string:find(InstanceId, ?TEST_ID_PREFIX),
case TestIdStart of
nomatch -> false;
_ -> string:equal(TestIdStart, InstanceId)
end.
conn_opts(#{authentication := none}) ->
#{};
conn_opts(#{authentication := #{username := Username, password := Password}}) ->
@ -297,7 +291,7 @@ replayq_dir(ClientId) ->
filename:join([emqx:data_dir(), "pulsar", emqx_utils_conv:bin(ClientId)]).
producer_name(InstanceId, ChannelId) ->
case is_dry_run(InstanceId) of
case emqx_resource:is_dry_run(InstanceId) of
%% do not create more atom
true ->
pulsar_producer_probe_worker;

2
apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq_connector.erl
View File

@ -31,6 +31,7 @@
on_remove_channel/3,
on_get_channels/1,
on_stop/2,
resource_type/0,
callback_mode/0,
on_get_status/2,
on_get_channel_status/3,
@ -60,6 +61,7 @@ fields(config) ->
%% ===================================================================
%% emqx_resource callback
resource_type() -> rabbitmq.
callback_mode() -> always_sync.

2
apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_redis, [
{description, "EMQX Enterprise Redis Bridge"},
{vsn, "0.1.8"},
{vsn, "0.1.9"},
{registered, []},
{applications, [
kernel,

5
apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl
View File

@ -12,6 +12,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_add_channel/4,
on_remove_channel/3,
@ -29,7 +30,9 @@
%% resource callbacks
%% -------------------------------------------------------------------------------------------------
callback_mode() -> always_sync.
resource_type() -> emqx_redis:resource_type().
callback_mode() -> emqx_redis:callback_mode().
on_add_channel(
_InstanceId,

2
apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_rocketmq, [
{description, "EMQX Enterprise RocketMQ Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [kernel, stdlib, emqx_resource, rocketmq]},
{env, [

3
apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq_connector.erl
View File

@ -16,6 +16,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -90,6 +91,8 @@ servers() ->
%% `emqx_resource' API
%%========================================================================================
resource_type() -> rocketmq.
callback_mode() -> always_sync.
on_start(

3
apps/emqx_bridge_s3/src/emqx_bridge_s3_connector.erl
View File

@ -13,6 +13,7 @@
-behaviour(emqx_resource).
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -92,6 +93,8 @@
-define(AGGREG_SUP, emqx_bridge_s3_sup).
%%
-spec resource_type() -> resource_type().
resource_type() -> s3.
-spec callback_mode() -> callback_mode().
callback_mode() ->

2
apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver_connector.erl
View File

@ -30,6 +30,7 @@
%% callbacks for behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -173,6 +174,7 @@ server() ->
%%====================================================================
%% Callbacks defined in emqx_resource
%%====================================================================
resource_type() -> sqlserver.
callback_mode() -> always_sync.

2
apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_syskeeper, [
{description, "EMQX Enterprise Data bridge for Syskeeper"},
{vsn, "0.1.3"},
{vsn, "0.1.4"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_connector.erl
View File

@ -18,6 +18,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -147,6 +148,7 @@ server() ->
%% -------------------------------------------------------------------------------------------------
%% `emqx_resource' API
resource_type() -> syskeeper.
callback_mode() -> always_sync.

3
apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_proxy_server.erl
View File

@ -12,6 +12,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
query_mode/1,
on_start/2,
on_stop/2,
@ -40,6 +41,8 @@
%% -------------------------------------------------------------------------------------------------
%% emqx_resource
resource_type() ->
syskeeper_proxy_server.
query_mode(_) ->
no_queries.

2
apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_tdengine, [
{description, "EMQX Enterprise TDEngine Bridge"},
{vsn, "0.2.1"},
{vsn, "0.2.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine_connector.erl
View File

@ -19,6 +19,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -140,6 +141,7 @@ connector_example_values() ->
%%========================================================================================
%% `emqx_resource' API
%%========================================================================================
resource_type() -> tdengine.
callback_mode() -> always_sync.

5
apps/emqx_conf/src/emqx_conf_schema_inject.erl
View File

@ -55,7 +55,10 @@ authn_mods(ce) ->
];
authn_mods(ee) ->
authn_mods(ce) ++
[emqx_gcp_device_authn_schema].
[
emqx_gcp_device_authn_schema,
emqx_authn_scram_http_schema
].
authz() ->
[{emqx_authz_schema, authz_mods()}].

2
apps/emqx_connector/include/emqx_connector.hrl
View File

@ -37,4 +37,4 @@
"The " ++ TYPE ++ " default port " ++ DEFAULT_PORT ++ " is used if `[:Port]` is not specified."
).
-define(CONNECTOR_RESOURCE_GROUP, <<"emqx_connector">>).
-define(CONNECTOR_RESOURCE_GROUP, <<"connector">>).

5
apps/emqx_connector/src/emqx_connector_resource.erl
View File

@ -18,6 +18,7 @@
-include("../../emqx_bridge/include/emqx_bridge_resource.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("emqx_resource/include/emqx_resource.hrl").
-include("emqx_connector.hrl").
-export([
connector_to_resource_type/1,
@ -126,7 +127,7 @@ create(Type, Name, Conf0, Opts) ->
Conf = Conf0#{connector_type => TypeBin, connector_name => Name},
{ok, _Data} = emqx_resource:create_local(
ResourceId,
<<"emqx_connector">>,
?CONNECTOR_RESOURCE_GROUP,
?MODULE:connector_to_resource_type(Type),
parse_confs(TypeBin, Name, Conf),
parse_opts(Conf, Opts)
@ -208,7 +209,7 @@ create_dry_run(Type, Conf) ->
create_dry_run(Type, Conf, fun(_) -> ok end).
create_dry_run(Type, Conf0, Callback) ->
%% Already typechecked, no need to catch errors
%% Already type checked, no need to catch errors
TypeBin = bin(Type),
TypeAtom = safe_atom(Type),
%% We use a fixed name here to avoid creating an atom

5
apps/emqx_connector/test/emqx_connector_SUITE.erl
View File

@ -50,6 +50,7 @@ t_connector_lifecycle({init, Config}) ->
meck:new(emqx_connector_resource, [passthrough]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
meck:new(?CONNECTOR, [non_strict]),
meck:expect(?CONNECTOR, resource_type, 0, dummy),
meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
meck:expect(?CONNECTOR, on_stop, 2, ok),
@ -171,6 +172,7 @@ t_remove_fail({'init', Config}) ->
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
meck:new(?CONNECTOR, [non_strict]),
meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
meck:expect(?CONNECTOR, resource_type, 0, dummy),
meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
meck:expect(?CONNECTOR, on_get_channels, 1, [{<<"my_channel">>, #{enable => true}}]),
meck:expect(?CONNECTOR, on_add_channel, 4, {ok, connector_state}),
@ -234,6 +236,7 @@ t_create_with_bad_name_direct_path({init, Config}) ->
meck:new(emqx_connector_resource, [passthrough]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
meck:new(?CONNECTOR, [non_strict]),
meck:expect(?CONNECTOR, resource_type, 0, dummy),
meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
meck:expect(?CONNECTOR, on_stop, 2, ok),
@ -265,6 +268,7 @@ t_create_with_bad_name_root_path({init, Config}) ->
meck:new(emqx_connector_resource, [passthrough]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
meck:new(?CONNECTOR, [non_strict]),
meck:expect(?CONNECTOR, resource_type, 0, dummy),
meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
meck:expect(?CONNECTOR, on_stop, 2, ok),
@ -299,6 +303,7 @@ t_no_buffer_workers({'init', Config}) ->
meck:new(emqx_connector_resource, [passthrough]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
meck:new(?CONNECTOR, [non_strict]),
meck:expect(?CONNECTOR, resource_type, 0, dummy),
meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
meck:expect(?CONNECTOR, on_get_channels, 1, []),

1
apps/emqx_connector/test/emqx_connector_api_SUITE.erl
View File

@ -224,6 +224,7 @@ init_mocks(_TestCase) ->
meck:new(emqx_connector_resource, [passthrough, no_link]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
meck:expect(?CONNECTOR_IMPL, resource_type, 0, dummy),
meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
meck:expect(
?CONNECTOR_IMPL,

3
apps/emqx_connector/test/emqx_connector_dummy_impl.erl
View File

@ -15,8 +15,10 @@
%% this module is only intended to be mocked
-module(emqx_connector_dummy_impl).
-behavior(emqx_resource).
-export([
resource_type/0,
query_mode/1,
callback_mode/0,
on_start/2,
@ -25,6 +27,7 @@
on_get_channel_status/3
]).
resource_type() -> dummy.
query_mode(_) -> error(unexpected).
callback_mode() -> error(unexpected).
on_start(_, _) -> error(unexpected).

2
apps/emqx_dashboard_sso/src/emqx_dashboard_sso_manager.erl
View File

@ -45,7 +45,7 @@
-define(MOD_TAB, emqx_dashboard_sso).
-define(MOD_KEY_PATH, [dashboard, sso]).
-define(MOD_KEY_PATH(Sub), [dashboard, sso, Sub]).
-define(RESOURCE_GROUP, <<"emqx_dashboard_sso">>).
-define(RESOURCE_GROUP, <<"dashboard_sso">>).
-define(NO_ERROR, <<>>).
-define(DEFAULT_RESOURCE_OPTS, #{
start_after_created => false

2
apps/emqx_dashboard_sso/test/emqx_dashboard_sso_ldap_SUITE.erl
View File

@ -24,7 +24,7 @@
-define(MOD_TAB, emqx_dashboard_sso).
-define(MOD_KEY_PATH, [dashboard, sso, ldap]).
-define(RESOURCE_GROUP, <<"emqx_dashboard_sso">>).
-define(RESOURCE_GROUP, <<"dashboard_sso">>).
-import(emqx_mgmt_api_test_util, [request/2, request/3, uri/1, request_api/3]).

5
apps/emqx_gateway/src/emqx_gateway_api_authn.erl
View File

@ -381,6 +381,9 @@ params_fuzzy_in_qs() ->
schema_authn() ->
emqx_dashboard_swagger:schema_with_examples(
emqx_authn_schema:authenticator_type_without([emqx_authn_scram_mnesia_schema]),
emqx_authn_schema:authenticator_type_without([
emqx_authn_scram_mnesia_schema,
emqx_authn_scram_http_schema
]),
emqx_authn_api:authenticator_examples()
).

2
apps/emqx_ldap/src/emqx_ldap.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_ldap, [
{description, "EMQX LDAP Connector"},
{vsn, "0.1.8"},
{vsn, "0.1.9"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_ldap/src/emqx_ldap.erl
View File

@ -27,6 +27,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -129,6 +130,8 @@ ensure_username(Field) ->
emqx_connector_schema_lib:username(Field).
%% ===================================================================
resource_type() -> ldap.
callback_mode() -> always_sync.
-spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

2
apps/emqx_mongodb/src/emqx_mongodb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_mongodb, [
{description, "EMQX MongoDB Connector"},
{vsn, "0.1.6"},
{vsn, "0.1.7"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_mongodb/src/emqx_mongodb.erl
View File

@ -26,6 +26,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -172,6 +173,7 @@ desc(_) ->
undefined.
%% ===================================================================
resource_type() -> mongodb.
callback_mode() -> always_sync.

2
apps/emqx_mysql/src/emqx_mysql.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_mysql, [
{description, "EMQX MySQL Database Connector"},
{vsn, "0.1.9"},
{vsn, "0.2.0"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_mysql/src/emqx_mysql.erl
View File

@ -25,6 +25,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -91,6 +92,8 @@ server() ->
emqx_schema:servers_sc(Meta, ?MYSQL_HOST_OPTIONS).
%% ===================================================================
resource_type() -> mysql.
callback_mode() -> always_sync.
-spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

2
apps/emqx_oracle/src/emqx_oracle.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_oracle, [
{description, "EMQX Enterprise Oracle Database Connector"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_oracle/src/emqx_oracle.erl
View File

@ -21,6 +21,7 @@
%% callbacks for behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -67,6 +68,8 @@
batch_params_tokens := params_tokens()
}.
resource_type() -> oracle.
% As ecpool is not monitoring the worker's PID when doing a handover_async, the
% request can be lost if worker crashes. Thus, it's better to force requests to
% be sync for now.

2
apps/emqx_postgresql/src/emqx_postgresql.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_postgresql, [
{description, "EMQX PostgreSQL Database Connector"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_postgresql/src/emqx_postgresql.erl
View File

@ -29,6 +29,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -120,6 +121,8 @@ adjust_fields(Fields) ->
).
%% ===================================================================
resource_type() -> pgsql.
callback_mode() -> always_sync.
-spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

3
apps/emqx_redis/src/emqx_redis.erl
View File

@ -28,6 +28,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -119,6 +120,8 @@ redis_type(Type) ->
desc => ?DESC(Type)
}}.
resource_type() -> redis.
callback_mode() -> always_sync.
on_start(InstId, Config0) ->

10
apps/emqx_resource/include/emqx_resource.hrl
View File

@ -23,7 +23,8 @@
%% remind us of that.
-define(rm_status_stopped, stopped).
-type resource_type() :: module().
-type resource_type() :: atom().
-type resource_module() :: module().
-type resource_id() :: binary().
-type channel_id() :: binary().
-type raw_resource_config() :: binary() | raw_term_resource_config().
@ -158,5 +159,12 @@
%% See `hocon_tconf`
-define(TEST_ID_PREFIX, "t_probe_").
-define(RES_METRICS, resource_metrics).
-define(LOG_LEVEL(_L_),
case _L_ of
true -> info;
false -> warning
end
).
-define(TAG, "RESOURCE").
-define(RESOURCE_ALLOCATION_TAB, emqx_resource_allocations).

64
apps/emqx_resource/src/emqx_resource.erl
View File

@ -39,12 +39,10 @@
-export([
%% store the config and start the instance
create_local/4,
create_local/5,
create_dry_run_local/2,
create_dry_run_local/3,
create_dry_run_local/4,
recreate_local/3,
recreate_local/4,
%% remove the config and stop the instance
remove_local/1,
@ -98,6 +96,7 @@
-export([
%% get the callback mode of a specific module
get_callback_mode/1,
get_resource_type/1,
%% start the instance
call_start/3,
%% verify if the resource is working normally
@ -140,6 +139,8 @@
validate_name/1
]).
-export([is_dry_run/1]).
-export_type([
query_mode/0,
resource_id/0,
@ -279,16 +280,10 @@ is_resource_mod(Module) ->
%% =================================================================================
%% APIs for resource instances
%% =================================================================================
-spec create_local(resource_id(), resource_group(), resource_type(), resource_config()) ->
{ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
create_local(ResId, Group, ResourceType, Config) ->
create_local(ResId, Group, ResourceType, Config, #{}).
-spec create_local(
resource_id(),
resource_group(),
resource_type(),
resource_module(),
resource_config(),
creation_opts()
) ->
@ -296,7 +291,7 @@ create_local(ResId, Group, ResourceType, Config) ->
create_local(ResId, Group, ResourceType, Config, Opts) ->
emqx_resource_manager:ensure_resource(ResId, Group, ResourceType, Config, Opts).
-spec create_dry_run_local(resource_type(), resource_config()) ->
-spec create_dry_run_local(resource_module(), resource_config()) ->
ok | {error, Reason :: term()}.
create_dry_run_local(ResourceType, Config) ->
emqx_resource_manager:create_dry_run(ResourceType, Config).
@ -304,19 +299,21 @@ create_dry_run_local(ResourceType, Config) ->
create_dry_run_local(ResId, ResourceType, Config) ->
emqx_resource_manager:create_dry_run(ResId, ResourceType, Config).
-spec create_dry_run_local(resource_id(), resource_type(), resource_config(), OnReadyCallback) ->
-spec create_dry_run_local(
resource_id(),
resource_module(),
resource_config(),
OnReadyCallback
) ->
ok | {error, Reason :: term()}
when
OnReadyCallback :: fun((resource_id()) -> ok | {error, Reason :: term()}).
create_dry_run_local(ResId, ResourceType, Config, OnReadyCallback) ->
emqx_resource_manager:create_dry_run(ResId, ResourceType, Config, OnReadyCallback).
-spec recreate_local(resource_id(), resource_type(), resource_config()) ->
{ok, resource_data()} | {error, Reason :: term()}.
recreate_local(ResId, ResourceType, Config) ->
recreate_local(ResId, ResourceType, Config, #{}).
-spec recreate_local(resource_id(), resource_type(), resource_config(), creation_opts()) ->
-spec recreate_local(
resource_id(), resource_module(), resource_config(), creation_opts()
) ->
{ok, resource_data()} | {error, Reason :: term()}.
recreate_local(ResId, ResourceType, Config, Opts) ->
emqx_resource_manager:recreate(ResId, ResourceType, Config, Opts).
@ -330,11 +327,15 @@ remove_local(ResId) ->
ok;
Error ->
%% Only log, the ResId worker is always removed in manager's remove action.
?SLOG(warning, #{
msg => "remove_local_resource_failed",
error => Error,
resource_id => ResId
}),
?SLOG(
warning,
#{
msg => "remove_resource_failed",
error => Error,
resource_id => ResId
},
#{tag => ?TAG}
),
ok
end.
@ -487,6 +488,10 @@ list_group_instances(Group) -> emqx_resource_manager:list_group(Group).
get_callback_mode(Mod) ->
Mod:callback_mode().
-spec get_resource_type(module()) -> resource_type().
get_resource_type(Mod) ->
Mod:resource_type().
-spec call_start(resource_id(), module(), resource_config()) ->
{ok, resource_state()} | {error, Reason :: term()}.
call_start(ResId, Mod, Config) ->
@ -599,7 +604,7 @@ query_mode(Mod, Config, Opts) ->
maps:get(query_mode, Opts, sync)
end.
-spec check_config(resource_type(), raw_resource_config()) ->
-spec check_config(resource_module(), raw_resource_config()) ->
{ok, resource_config()} | {error, term()}.
check_config(ResourceType, Conf) ->
emqx_hocon:check(ResourceType, Conf).
@ -607,7 +612,7 @@ check_config(ResourceType, Conf) ->
-spec check_and_create_local(
resource_id(),
resource_group(),
resource_type(),
resource_module(),
raw_resource_config()
) ->
{ok, resource_data()} | {error, term()}.
@ -617,7 +622,7 @@ check_and_create_local(ResId, Group, ResourceType, RawConfig) ->
-spec check_and_create_local(
resource_id(),
resource_group(),
resource_type(),
resource_module(),
raw_resource_config(),
creation_opts()
) -> {ok, resource_data()} | {error, term()}.
@ -630,7 +635,7 @@ check_and_create_local(ResId, Group, ResourceType, RawConfig, Opts) ->
-spec check_and_recreate_local(
resource_id(),
resource_type(),
resource_module(),
raw_resource_config(),
creation_opts()
) ->
@ -769,6 +774,13 @@ validate_name(Name) ->
_ = validate_name(Name, #{atom_name => false}),
ok.
-spec is_dry_run(resource_id()) -> boolean().
is_dry_run(ResId) ->
case string:find(ResId, ?TEST_ID_PREFIX) of
nomatch -> false;
TestIdStart -> string:equal(TestIdStart, ResId)
end.
validate_name(<<>>, _Opts) ->
invalid_data("Name cannot be empty string");
validate_name(Name, _Opts) when size(Name) >= 255 ->

210
apps/emqx_resource/src/emqx_resource_manager.erl
View File

@ -75,6 +75,7 @@
-record(data, {
id,
group,
type,
mod,
callback_mode,
query_mode,
@ -166,7 +167,7 @@ where(ResId) ->
-spec ensure_resource(
resource_id(),
resource_group(),
resource_type(),
resource_module(),
resource_config(),
creation_opts()
) -> {ok, resource_data()}.
@ -179,7 +180,9 @@ ensure_resource(ResId, Group, ResourceType, Config, Opts) ->
end.
%% @doc Called from emqx_resource when recreating a resource which may or may not exist
-spec recreate(resource_id(), resource_type(), resource_config(), creation_opts()) ->
-spec recreate(
resource_id(), resource_module(), resource_config(), creation_opts()
) ->
{ok, resource_data()} | {error, not_found} | {error, updating_to_incorrect_resource_type}.
recreate(ResId, ResourceType, NewConfig, Opts) ->
case lookup(ResId) of
@ -222,8 +225,8 @@ create(ResId, Group, ResourceType, Config, Opts) ->
%% @doc Called from `emqx_resource` when doing a dry run for creating a resource instance.
%%
%% Triggers the `emqx_resource_manager_sup` supervisor to actually create
%% and link the process itself if not already started, and then immedately stops.
-spec create_dry_run(resource_type(), resource_config()) ->
%% and link the process itself if not already started, and then immediately stops.
-spec create_dry_run(resource_module(), resource_config()) ->
ok | {error, Reason :: term()}.
create_dry_run(ResourceType, Config) ->
ResId = make_test_id(),
@ -235,7 +238,9 @@ create_dry_run(ResId, ResourceType, Config) ->
do_nothing_on_ready(_ResId) ->
ok.
-spec create_dry_run(resource_id(), resource_type(), resource_config(), OnReadyCallback) ->
-spec create_dry_run(
resource_id(), resource_module(), resource_config(), OnReadyCallback
) ->
ok | {error, Reason :: term()}
when
OnReadyCallback :: fun((resource_id()) -> ok | {error, Reason :: term()}).
@ -245,7 +250,9 @@ create_dry_run(ResId, ResourceType, Config, OnReadyCallback) ->
true -> maps:get(resource_opts, Config, #{});
false -> #{}
end,
ok = emqx_resource_manager_sup:ensure_child(ResId, <<"dry_run">>, ResourceType, Config, Opts),
ok = emqx_resource_manager_sup:ensure_child(
ResId, <<"dry_run">>, ResourceType, Config, Opts
),
HealthCheckInterval = maps:get(health_check_interval, Opts, ?HEALTHCHECK_INTERVAL),
Timeout = emqx_utils:clamp(HealthCheckInterval, 5_000, 60_000),
case wait_for_ready(ResId, Timeout) of
@ -526,6 +533,7 @@ start_link(ResId, Group, ResourceType, Config, Opts) ->
),
Data = #data{
id = ResId,
type = emqx_resource:get_resource_type(ResourceType),
group = Group,
mod = ResourceType,
callback_mode = emqx_resource:get_callback_mode(ResourceType),
@ -710,11 +718,13 @@ handle_event(EventType, EventData, State, Data) ->
error,
#{
msg => "ignore_all_other_events",
resource_id => Data#data.id,
event_type => EventType,
event_data => EventData,
state => State,
data => emqx_utils:redact(Data)
}
},
#{tag => tag(Data#data.group, Data#data.type)}
),
keep_state_and_data.
@ -779,7 +789,8 @@ handle_remove_event(From, ClearMetrics, Data) ->
start_resource(Data, From) ->
%% in case the emqx_resource:call_start/2 hangs, the lookup/1 can read status from the cache
case emqx_resource:call_start(Data#data.id, Data#data.mod, Data#data.config) of
#data{id = ResId, mod = Mod, config = Config, group = Group, type = Type} = Data,
case emqx_resource:call_start(ResId, Mod, Config) of
{ok, ResourceState} ->
UpdatedData1 = Data#data{status = ?status_connecting, state = ResourceState},
%% Perform an initial health_check immediately before transitioning into a connected state
@ -787,12 +798,17 @@ start_resource(Data, From) ->
Actions = maybe_reply([{state_timeout, 0, health_check}], From, ok),
{next_state, ?state_connecting, update_state(UpdatedData2, Data), Actions};
{error, Reason} = Err ->
?SLOG(warning, #{
msg => "start_resource_failed",
id => Data#data.id,
reason => Reason
}),
_ = maybe_alarm(?status_disconnected, Data#data.id, Err, Data#data.error),
IsDryRun = emqx_resource:is_dry_run(ResId),
?SLOG(
log_level(IsDryRun),
#{
msg => "start_resource_failed",
resource_id => ResId,
reason => Reason
},
#{tag => tag(Group, Type)}
),
_ = maybe_alarm(?status_disconnected, IsDryRun, ResId, Err, Data#data.error),
%% Add channels and raise alarms
NewData1 = channels_health_check(?status_disconnected, add_channels(Data)),
%% Keep track of the error reason why the connection did not work
@ -823,13 +839,20 @@ add_channels(Data) ->
add_channels_in_list([], Data) ->
Data;
add_channels_in_list([{ChannelID, ChannelConfig} | Rest], Data) ->
#data{
id = ResId,
mod = Mod,
state = State,
added_channels = AddedChannelsMap,
group = Group,
type = Type
} = Data,
case
emqx_resource:call_add_channel(
Data#data.id, Data#data.mod, Data#data.state, ChannelID, ChannelConfig
ResId, Mod, State, ChannelID, ChannelConfig
)
of
{ok, NewState} ->
AddedChannelsMap = Data#data.added_channels,
%% Set the channel status to connecting to indicate that
%% we have not yet performed the initial health_check
NewAddedChannelsMap = maps:put(
@ -843,12 +866,17 @@ add_channels_in_list([{ChannelID, ChannelConfig} | Rest], Data) ->
},
add_channels_in_list(Rest, NewData);
{error, Reason} = Error ->
?SLOG(warning, #{
msg => add_channel_failed,
id => Data#data.id,
channel_id => ChannelID,
reason => Reason
}),
IsDryRun = emqx_resource:is_dry_run(ResId),
?SLOG(
log_level(IsDryRun),
#{
msg => "add_channel_failed",
resource_id => ResId,
channel_id => ChannelID,
reason => Reason
},
#{tag => tag(Group, Type)}
),
AddedChannelsMap = Data#data.added_channels,
NewAddedChannelsMap = maps:put(
ChannelID,
@ -859,7 +887,7 @@ add_channels_in_list([{ChannelID, ChannelConfig} | Rest], Data) ->
added_channels = NewAddedChannelsMap
},
%% Raise an alarm since the channel could not be added
_ = maybe_alarm(?status_disconnected, ChannelID, Error, no_prev_error),
_ = maybe_alarm(?status_disconnected, IsDryRun, ChannelID, Error, no_prev_error),
add_channels_in_list(Rest, NewData)
end.
@ -883,7 +911,8 @@ stop_resource(#data{id = ResId} = Data) ->
false ->
ok
end,
_ = maybe_clear_alarm(ResId),
IsDryRun = emqx_resource:is_dry_run(ResId),
_ = maybe_clear_alarm(IsDryRun, ResId),
ok = emqx_metrics_worker:reset_metrics(?RES_METRICS, ResId),
NewData#data{status = ?rm_status_stopped}.
@ -894,16 +923,24 @@ remove_channels(Data) ->
remove_channels_in_list([], Data, _KeepInChannelMap) ->
Data;
remove_channels_in_list([ChannelID | Rest], Data, KeepInChannelMap) ->
AddedChannelsMap = Data#data.added_channels,
#data{
id = ResId,
added_channels = AddedChannelsMap,
mod = Mod,
state = State,
group = Group,
type = Type
} = Data,
IsDryRun = emqx_resource:is_dry_run(ResId),
NewAddedChannelsMap =
case KeepInChannelMap of
true ->
AddedChannelsMap;
false ->
_ = maybe_clear_alarm(ChannelID),
_ = maybe_clear_alarm(IsDryRun, ChannelID),
maps:remove(ChannelID, AddedChannelsMap)
end,
case safe_call_remove_channel(Data#data.id, Data#data.mod, Data#data.state, ChannelID) of
case safe_call_remove_channel(ResId, Mod, State, ChannelID) of
{ok, NewState} ->
NewData = Data#data{
state = NewState,
@ -911,12 +948,18 @@ remove_channels_in_list([ChannelID | Rest], Data, KeepInChannelMap) ->
},
remove_channels_in_list(Rest, NewData, KeepInChannelMap);
{error, Reason} ->
?SLOG(warning, #{
msg => remove_channel_failed,
id => Data#data.id,
channel_id => ChannelID,
reason => Reason
}),
?SLOG(
log_level(IsDryRun),
#{
msg => "remove_channel_failed",
resource_id => ResId,
group => Group,
type => Type,
channel_id => ChannelID,
reason => Reason
},
#{tag => tag(Group, Type)}
),
NewData = Data#data{
added_channels = NewAddedChannelsMap
},
@ -995,8 +1038,8 @@ handle_not_connected_add_channel(From, ChannelId, ChannelConfig, State, Data) ->
handle_remove_channel(From, ChannelId, Data) ->
Channels = Data#data.added_channels,
%% Deactivate alarm
_ = maybe_clear_alarm(ChannelId),
IsDryRun = emqx_resource:is_dry_run(Data#data.id),
_ = maybe_clear_alarm(IsDryRun, ChannelId),
case
channel_status_is_channel_added(
maps:get(ChannelId, Channels, channel_status_not_added(undefined))
@ -1017,13 +1060,18 @@ handle_remove_channel(From, ChannelId, Data) ->
end.
handle_remove_channel_exists(From, ChannelId, Data) ->
#data{
id = Id,
group = Group,
type = Type,
added_channels = AddedChannelsMap
} = Data,
case
emqx_resource:call_remove_channel(
Data#data.id, Data#data.mod, Data#data.state, ChannelId
Id, Data#data.mod, Data#data.state, ChannelId
)
of
{ok, NewState} ->
AddedChannelsMap = Data#data.added_channels,
NewAddedChannelsMap = maps:remove(ChannelId, AddedChannelsMap),
UpdatedData = Data#data{
state = NewState,
@ -1031,13 +1079,17 @@ handle_remove_channel_exists(From, ChannelId, Data) ->
},
{keep_state, update_state(UpdatedData, Data), [{reply, From, ok}]};
{error, Reason} = Error ->
%% Log the error as a warning
?SLOG(warning, #{
msg => remove_channel_failed,
id => Data#data.id,
channel_id => ChannelId,
reason => Reason
}),
IsDryRun = emqx_resource:is_dry_run(Id),
?SLOG(
log_level(IsDryRun),
#{
msg => "remove_channel_failed",
resource_id => Id,
channel_id => ChannelId,
reason => Reason
},
#{tag => tag(Group, Type)}
),
{keep_state_and_data, [{reply, From, Error}]}
end.
@ -1048,7 +1100,8 @@ handle_not_connected_and_not_connecting_remove_channel(From, ChannelId, Data) ->
Channels = Data#data.added_channels,
NewChannels = maps:remove(ChannelId, Channels),
NewData = Data#data{added_channels = NewChannels},
_ = maybe_clear_alarm(ChannelId),
IsDryRun = emqx_resource:is_dry_run(Data#data.id),
_ = maybe_clear_alarm(IsDryRun, ChannelId),
{keep_state, update_state(NewData, Data), [{reply, From, ok}]}.
handle_manual_resource_health_check(From, Data0 = #data{hc_workers = #{resource := HCWorkers}}) when
@ -1117,7 +1170,8 @@ continue_with_health_check(#data{} = Data0, CurrentState, HCRes) ->
error = PrevError
} = Data0,
{NewStatus, NewState, Err} = parse_health_check_result(HCRes, Data0),
_ = maybe_alarm(NewStatus, ResId, Err, PrevError),
IsDryRun = emqx_resource:is_dry_run(ResId),
_ = maybe_alarm(NewStatus, IsDryRun, ResId, Err, PrevError),
ok = maybe_resume_resource_workers(ResId, NewStatus),
Data1 = Data0#data{
state = NewState, status = NewStatus, error = Err
@ -1141,11 +1195,17 @@ continue_resource_health_check_connected(NewStatus, Data0) ->
Actions = Replies ++ resource_health_check_actions(Data),
{keep_state, Data, Actions};
_ ->
?SLOG(warning, #{
msg => "health_check_failed",
id => Data0#data.id,
status => NewStatus
}),
#data{id = ResId, group = Group, type = Type} = Data0,
IsDryRun = emqx_resource:is_dry_run(ResId),
?SLOG(
log_level(IsDryRun),
#{
msg => "health_check_failed",
resource_id => ResId,
status => NewStatus
},
#{tag => tag(Group, Type)}
),
%% Note: works because, coincidentally, channel/resource status is a
%% subset of resource manager state... But there should be a conversion
%% between the two here, as resource manager also has `stopped', which is
@ -1241,7 +1301,7 @@ channels_health_check(?status_connected = _ConnectorStatus, Data0) ->
channels_health_check(?status_connecting = _ConnectorStatus, Data0) ->
%% Whenever the resource is connecting:
%% 1. Change the status of all added channels to connecting
%% 2. Raise alarms (TODO: if it is a probe we should not raise alarms)
%% 2. Raise alarms
Channels = Data0#data.added_channels,
ChannelsToChangeStatusFor = [
{ChannelId, Config}
@ -1267,9 +1327,10 @@ channels_health_check(?status_connecting = _ConnectorStatus, Data0) ->
|| {ChannelId, NewStatus} <- maps:to_list(NewChannels)
],
%% Raise alarms for all channels
IsDryRun = emqx_resource:is_dry_run(Data0#data.id),
lists:foreach(
fun({ChannelId, Status, PrevStatus}) ->
maybe_alarm(?status_connecting, ChannelId, Status, PrevStatus)
maybe_alarm(?status_connecting, IsDryRun, ChannelId, Status, PrevStatus)
end,
ChannelsWithNewAndPrevErrorStatuses
),
@ -1302,9 +1363,10 @@ channels_health_check(ConnectorStatus, Data0) ->
|| {ChannelId, #{config := Config} = OldStatus} <- maps:to_list(Data1#data.added_channels)
],
%% Raise alarms
IsDryRun = emqx_resource:is_dry_run(Data1#data.id),
_ = lists:foreach(
fun({ChannelId, OldStatus, NewStatus}) ->
_ = maybe_alarm(NewStatus, ChannelId, NewStatus, OldStatus)
_ = maybe_alarm(NewStatus, IsDryRun, ChannelId, NewStatus, OldStatus)
end,
ChannelsWithNewAndOldStatuses
),
@ -1413,13 +1475,14 @@ continue_channel_health_check_connected_no_update_during_check(ChannelId, OldSta
NewStatus = maps:get(ChannelId, Data1#data.added_channels),
ChannelsToRemove = [ChannelId || not channel_status_is_channel_added(NewStatus)],
Data = remove_channels_in_list(ChannelsToRemove, Data1, true),
IsDryRun = emqx_resource:is_dry_run(Data1#data.id),
%% Raise/clear alarms
case NewStatus of
#{status := ?status_connected} ->
_ = maybe_clear_alarm(ChannelId),
_ = maybe_clear_alarm(IsDryRun, ChannelId),
ok;
_ ->
_ = maybe_alarm(NewStatus, ChannelId, NewStatus, OldStatus),
_ = maybe_alarm(NewStatus, IsDryRun, ChannelId, NewStatus, OldStatus),
ok
end,
Data.
@ -1583,15 +1646,21 @@ remove_runtime_data(#data{} = Data0) ->
health_check_interval(Opts) ->
maps:get(health_check_interval, Opts, ?HEALTHCHECK_INTERVAL).
-spec maybe_alarm(resource_status(), resource_id(), _Error :: term(), _PrevError :: term()) -> ok.
maybe_alarm(?status_connected, _ResId, _Error, _PrevError) ->
-spec maybe_alarm(
resource_status(),
boolean(),
resource_id(),
_Error :: term(),
_PrevError :: term()
) -> ok.
maybe_alarm(?status_connected, _IsDryRun, _ResId, _Error, _PrevError) ->
ok;
maybe_alarm(_Status, <<?TEST_ID_PREFIX, _/binary>>, _Error, _PrevError) ->
maybe_alarm(_Status, true, _ResId, _Error, _PrevError) ->
ok;
%% Assume that alarm is already active
maybe_alarm(_Status, _ResId, Error, Error) ->
maybe_alarm(_Status, _IsDryRun, _ResId, Error, Error) ->
ok;
maybe_alarm(_Status, ResId, Error, _PrevError) ->
maybe_alarm(_Status, false, ResId, Error, _PrevError) ->
HrError =
case Error of
{error, undefined} ->
@ -1623,10 +1692,10 @@ maybe_resume_resource_workers(ResId, ?status_connected) ->
maybe_resume_resource_workers(_, _) ->
ok.
-spec maybe_clear_alarm(resource_id()) -> ok | {error, not_found}.
maybe_clear_alarm(<<?TEST_ID_PREFIX, _/binary>>) ->
-spec maybe_clear_alarm(boolean(), resource_id()) -> ok | {error, not_found}.
maybe_clear_alarm(true, _ResId) ->
ok;
maybe_clear_alarm(ResId) ->
maybe_clear_alarm(false, ResId) ->
emqx_alarm:safe_deactivate(ResId).
parse_health_check_result(Status, Data) when ?IS_STATUS(Status) ->
@ -1642,7 +1711,8 @@ parse_health_check_result({error, Error}, Data) ->
msg => "health_check_exception",
resource_id => Data#data.id,
reason => Error
}
},
#{tag => tag(Data#data.group, Data#data.type)}
),
{?status_disconnected, Data#data.state, {error, Error}}.
@ -1794,10 +1864,18 @@ add_or_update_channel_status(Data, ChannelId, ChannelConfig, State) ->
ChannelStatus = channel_status({error, resource_not_operational}, ChannelConfig),
NewChannels = maps:put(ChannelId, ChannelStatus, Channels),
ResStatus = state_to_status(State),
maybe_alarm(ResStatus, ChannelId, ChannelStatus, no_prev),
IsDryRun = emqx_resource:is_dry_run(ChannelId),
maybe_alarm(ResStatus, IsDryRun, ChannelId, ChannelStatus, no_prev),
Data#data{added_channels = NewChannels}.
state_to_status(?state_stopped) -> ?rm_status_stopped;
state_to_status(?state_connected) -> ?status_connected;
state_to_status(?state_connecting) -> ?status_connecting;
state_to_status(?state_disconnected) -> ?status_disconnected.
log_level(true) -> info;
log_level(false) -> warning.
tag(Group, Type) ->
Str = emqx_utils_conv:str(Group) ++ "/" ++ emqx_utils_conv:str(Type),
string:uppercase(Str).

5
apps/emqx_resource/src/emqx_resource_manager_sup.erl
View File

@ -58,10 +58,11 @@ init([]) ->
child_spec(ResId, Group, ResourceType, Config, Opts) ->
#{
id => ResId,
start => {emqx_resource_manager, start_link, [ResId, Group, ResourceType, Config, Opts]},
start =>
{emqx_resource_manager, start_link, [ResId, Group, ResourceType, Config, Opts]},
restart => transient,
%% never force kill a resource manager.
%% becasue otherwise it may lead to release leak,
%% because otherwise it may lead to release leak,
%% resource_manager's terminate callback calls resource on_stop
shutdown => infinity,
type => worker,

46
apps/emqx_resource/src/emqx_resource_metrics.erl
View File

@ -17,6 +17,7 @@
-module(emqx_resource_metrics).
-include_lib("emqx/include/logger.hrl").
-include("emqx_resource.hrl").
-export([
events/0,
@ -74,7 +75,6 @@
success_get/1
]).
-define(RES_METRICS, resource_metrics).
-define(TELEMETRY_PREFIX, emqx, resource).
-spec events() -> [telemetry:event_name()].
@ -127,15 +127,19 @@ handle_telemetry_event(
%% We catch errors to avoid detaching the telemetry handler function.
%% When restarting a resource while it's under load, there might be transient
%% failures while the metrics are not yet created.
?SLOG(warning, #{
msg => "handle_resource_metrics_failed",
hint => "transient failures may occur when restarting a resource",
kind => Kind,
reason => Reason,
stacktrace => Stacktrace,
resource_id => ID,
event => Event
}),
?SLOG(
warning,
#{
msg => "handle_resource_metrics_failed",
hint => "transient failures may occur when restarting a resource",
kind => Kind,
reason => Reason,
stacktrace => Stacktrace,
resource_id => ID,
event => Event
},
#{tag => ?TAG}
),
ok
end;
handle_telemetry_event(
@ -151,15 +155,19 @@ handle_telemetry_event(
%% We catch errors to avoid detaching the telemetry handler function.
%% When restarting a resource while it's under load, there might be transient
%% failures while the metrics are not yet created.
?SLOG(warning, #{
msg => "handle_resource_metrics_failed",
hint => "transient failures may occur when restarting a resource",
kind => Kind,
reason => Reason,
stacktrace => Stacktrace,
resource_id => ID,
event => Event
}),
?SLOG(
warning,
#{
msg => "handle_resource_metrics_failed",
hint => "transient failures may occur when restarting a resource",
kind => Kind,
reason => Reason,
stacktrace => Stacktrace,
resource_id => ID,
event => Event
},
#{tag => ?TAG}
),
ok
end;
handle_telemetry_event(_EventName, _Measurements, _Metadata, _HandlerConfig) ->

35
apps/emqx_resource/src/emqx_resource_pool.erl
View File

@ -26,6 +26,7 @@
]).
-include_lib("emqx/include/logger.hrl").
-include("emqx_resource.hrl").
-ifndef(TEST).
-define(HEALTH_CHECK_TIMEOUT, 15000).
@ -37,33 +38,43 @@
start(Name, Mod, Options) ->
case ecpool:start_sup_pool(Name, Mod, Options) of
{ok, _} ->
?SLOG(info, #{msg => "start_ecpool_ok", pool_name => Name}),
?SLOG(info, #{msg => "start_ecpool_ok", pool_name => Name}, #{tag => ?TAG}),
ok;
{error, {already_started, _Pid}} ->
stop(Name),
start(Name, Mod, Options);
{error, Reason} ->
NReason = parse_reason(Reason),
?SLOG(error, #{
msg => "start_ecpool_error",
pool_name => Name,
reason => NReason
}),
IsDryRun = emqx_resource:is_dry_run(Name),
?SLOG(
?LOG_LEVEL(IsDryRun),
#{
msg => "start_ecpool_error",
resource_id => Name,
reason => NReason
},
#{tag => ?TAG}
),
{error, {start_pool_failed, Name, NReason}}
end.
stop(Name) ->
case ecpool:stop_sup_pool(Name) of
ok ->
?SLOG(info, #{msg => "stop_ecpool_ok", pool_name => Name});
?SLOG(info, #{msg => "stop_ecpool_ok", pool_name => Name}, #{tag => ?TAG});
{error, not_found} ->
ok;
{error, Reason} ->
?SLOG(error, #{
msg => "stop_ecpool_failed",
pool_name => Name,
reason => Reason
}),
IsDryRun = emqx_resource:is_dry_run(Name),
?SLOG(
?LOG_LEVEL(IsDryRun),
#{
msg => "stop_ecpool_failed",
resource_id => Name,
reason => Reason
},
#{tag => ?TAG}
),
error({stop_pool_failed, Name, Reason})
end.

6
apps/emqx_resource/src/proto/emqx_resource_proto_v1.erl
View File

@ -40,7 +40,7 @@ deprecated_since() ->
-spec create(
resource_id(),
resource_group(),
resource_type(),
resource_module(),
resource_config(),
creation_opts()
) ->
@ -51,7 +51,7 @@ create(ResId, Group, ResourceType, Config, Opts) ->
]).
-spec create_dry_run(
resource_type(),
resource_module(),
resource_config()
) ->
ok | {error, Reason :: term()}.
@ -60,7 +60,7 @@ create_dry_run(ResourceType, Config) ->
-spec recreate(
resource_id(),
resource_type(),
resource_module(),
resource_config(),
creation_opts()
) ->

3
apps/emqx_resource/test/emqx_connector_demo.erl
View File

@ -24,6 +24,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -62,6 +63,8 @@ register(required) -> true;
register(default) -> false;
register(_) -> undefined.
resource_type() -> demo.
callback_mode() ->
persistent_term:get(?CM_KEY).

Some files were not shown because too many files have changed in this diff Show More