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

Merge pull request #12020 from thalesmg/sync-r53-m-20231124 

chore: sync `release-53` to `master`
This commit is contained in:
zhongwencool 2023-11-24 22:54:53 +08:00 committed by GitHub
parent 2d4b9a7b9d 261fe8a831
commit 8f548f4cbb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 478 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
apps/emqx_auth/src/emqx_authn/emqx_authn_api.erl
View File

@ -1111,10 +1111,7 @@ list_users(ChainName, AuthenticatorID, QueryString) ->
{error, page_limit_invalid} ->
{400, #{code => <<"INVALID_PARAMETER">>, message => <<"page_limit_invalid">>}};
{error, Reason} ->
{400, #{
code => <<"INVALID_PARAMETER">>,
message => list_to_binary(io_lib:format("Reason ~p", [Reason]))
}};
serialize_error({user_error, Reason});
Result ->
{200, Result}
end.
@ -1176,6 +1173,16 @@ serialize_error({user_error, not_found}) ->
code => <<"NOT_FOUND">>,
message => binfmt("User not found", [])
}};
serialize_error({user_error, {not_found, {chain, ?GLOBAL}}}) ->
{404, #{
code => <<"NOT_FOUND">>,
message => <<"Authenticator not found in the 'global' scope">>
}};
serialize_error({user_error, {not_found, {chain, Name}}}) ->
{400, #{
code => <<"BAD_REQUEST">>,
message => binfmt("No authentication has been created for listener ~p", [Name])
}};
serialize_error({user_error, already_exist}) ->
{409, #{
code => <<"ALREADY_EXISTS">>,

2
apps/emqx_auth/src/emqx_authn/emqx_authn_password_hashing.erl
View File

@ -67,7 +67,7 @@
-define(SALT_ROUNDS_MIN, 5).
-define(SALT_ROUNDS_MAX, 10).
namespace() -> "authn-hash".
namespace() -> "authn_hash".
roots() -> [pbkdf2, bcrypt, bcrypt_rw, bcrypt_rw_api, simple].
fields(bcrypt_rw) ->

3
apps/emqx_auth/src/emqx_authz/sources/emqx_authz_file_schema.erl
View File

@ -22,6 +22,7 @@
-behaviour(emqx_authz_schema).
-export([
namespace/0,
type/0,
fields/1,
desc/1,
@ -30,6 +31,8 @@
select_union_member/1
]).
namespace() -> "authz".
type() -> ?AUTHZ_TYPE.
fields(file) ->

13
apps/emqx_auth/test/emqx_authn/emqx_authn_api_SUITE.erl
View File

@ -435,6 +435,19 @@ test_authenticator_position(PathPrefix) ->
PathPrefix ++ [?CONF_NS]
).
t_authenticator_users_not_found(_) ->
GlobalUser = #{user_id => <<"global_user">>, password => <<"p1">>},
{ok, 404, _} = request(
get,
uri([?CONF_NS, "password_based:built_in_database", "users"])
),
{ok, 404, _} = request(
post,
uri([?CONF_NS, "password_based:built_in_database", "users"]),
GlobalUser
),
ok.
%% listener authn api is not supported since 5.1.0
%% Don't support listener switch to global chain.
ignore_switch_to_global_chain(_) ->

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

@ -54,7 +54,7 @@ t_check_schema(_Config) ->
?assertThrow(
#{
path := "authentication.1.password_hash_algorithm.name",
matched_type := "authn:builtin_db/authn-hash:simple",
matched_type := "authn:builtin_db/authn_hash:simple",
reason := unable_to_convert_to_enum_symbol
},
Check(ConfigNotOk)

38
apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl
View File

@ -34,16 +34,22 @@ init_per_suite(Config) ->
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
case emqx_common_test_helpers:is_tcp_server_available(GCPEmulatorHost, GCPEmulatorPort) of
true ->
ok = emqx_common_test_helpers:start_apps([emqx_conf]),
ok = emqx_connector_test_helpers:start_apps([
emqx_resource, emqx_bridge, emqx_rule_engine
]),
{ok, _} = application:ensure_all_started(emqx_connector),
Apps = emqx_cth_suite:start(
[
emqx,
emqx_conf,
emqx_bridge_gcp_pubsub,
emqx_bridge,
emqx_rule_engine
],
#{work_dir => emqx_cth_suite:work_dir(Config)}
),
emqx_mgmt_api_test_util:init_suite(),
HostPort = GCPEmulatorHost ++ ":" ++ GCPEmulatorPortStr,
true = os:putenv("PUBSUB_EMULATOR_HOST", HostPort),
Client = start_control_client(),
[
{apps, Apps},
{proxy_name, ProxyName},
{proxy_host, ProxyHost},
{proxy_port, ProxyPort},
@ -62,12 +68,11 @@ init_per_suite(Config) ->
end.
end_per_suite(Config) ->
Apps = ?config(apps, Config),
Client = ?config(client, Config),
stop_control_client(Client),
emqx_mgmt_api_test_util:end_suite(),
ok = emqx_common_test_helpers:stop_apps([emqx_conf]),
ok = emqx_connector_test_helpers:stop_apps([emqx_bridge, emqx_resource, emqx_rule_engine]),
_ = application:stop(emqx_connector),
emqx_cth_suite:stop(Apps),
os:unsetenv("PUBSUB_EMULATOR_HOST"),
ok.
@ -1472,7 +1477,7 @@ t_pull_worker_death(Config) ->
[PullWorkerPid | _] = get_pull_worker_pids(Config),
Ref = monitor(process, PullWorkerPid),
sys:terminate(PullWorkerPid, die),
sys:terminate(PullWorkerPid, die, 20_000),
receive
{'DOWN', Ref, process, PullWorkerPid, _} ->
ok
@ -1494,10 +1499,11 @@ t_pull_worker_death(Config) ->
ok.
t_async_worker_death_mid_pull(Config) ->
ct:timetrap({seconds, 120}),
ct:timetrap({seconds, 122}),
[#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
Payload = emqx_guid:to_hexstr(emqx_guid:gen()),
?check_trace(
#{timetrap => 120_000},
begin
start_and_subscribe_mqtt(Config),
@ -1513,23 +1519,28 @@ t_async_worker_death_mid_pull(Config) ->
#{?snk_kind := gcp_pubsub_consumer_worker_reply_delegator}
),
spawn_link(fun() ->
ct:pal("will kill async workers"),
?tp_span(
kill_async_worker,
#{},
begin
%% produce a message while worker is being killed
Messages = [#{<<"data">> => Payload}],
ct:pal("publishing message"),
pubsub_publish(Config, PubSubTopic, Messages),
ct:pal("published message"),
AsyncWorkerPids = get_async_worker_pids(Config),
emqx_utils:pmap(
fun(AsyncWorkerPid) ->
Ref = monitor(process, AsyncWorkerPid),
sys:terminate(AsyncWorkerPid, die),
ct:pal("killing pid ~p", [AsyncWorkerPid]),
sys:terminate(AsyncWorkerPid, die, 20_000),
receive
{'DOWN', Ref, process, AsyncWorkerPid, _} ->
ct:pal("killed pid ~p", [AsyncWorkerPid]),
ok
after 500 -> ct:fail("async worker didn't die")
after 500 -> ct:fail("async worker ~p didn't die", [AsyncWorkerPid])
end,
ok
end,
@ -1538,7 +1549,8 @@ t_async_worker_death_mid_pull(Config) ->
ok
end
)
),
ct:pal("killed async workers")
end),
?assertMatch(

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

@ -54,7 +54,7 @@
%%=====================================================================
%% Hocon schema
namespace() -> "connector-http".
namespace() -> "connector_http".
roots() ->
fields(config).

2
apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector_schema.erl
View File

@ -36,7 +36,7 @@
-define(MQTT_HOST_OPTS, #{default_port => 1883}).
namespace() -> "connector-mqtt".
namespace() -> "connector_mqtt".
roots() ->
fields("config").

271
apps/emqx_conf/src/emqx_conf.erl
View File

@ -28,7 +28,7 @@
-export([remove/2, remove/3]).
-export([tombstone/2]).
-export([reset/2, reset/3]).
-export([dump_schema/2]).
-export([dump_schema/2, reformat_schema_dump/1]).
-export([schema_module/0]).
%% TODO: move to emqx_dashboard when we stop building api schema at build time
@ -180,9 +180,263 @@ gen_schema_json(Dir, SchemaModule, Lang) ->
include_importance_up_from => IncludeImportance,
desc_resolver => make_desc_resolver(Lang)
},
JsonMap = hocon_schema_json:gen(SchemaModule, Opts),
IoData = emqx_utils_json:encode(JsonMap, [pretty, force_utf8]),
ok = file:write_file(SchemaJsonFile, IoData).
StructsJsonArray = hocon_schema_json:gen(SchemaModule, Opts),
IoData = emqx_utils_json:encode(StructsJsonArray, [pretty, force_utf8]),
ok = file:write_file(SchemaJsonFile, IoData),
ok = gen_preformat_md_json_files(Dir, StructsJsonArray, Lang).
gen_preformat_md_json_files(Dir, StructsJsonArray, Lang) ->
NestedStruct = reformat_schema_dump(StructsJsonArray),
%% write to files
NestedJsonFile = filename:join([Dir, "schmea-v2-" ++ Lang ++ ".json"]),
io:format(user, "===< Generating: ~s~n", [NestedJsonFile]),
ok = file:write_file(
NestedJsonFile, emqx_utils_json:encode(NestedStruct, [pretty, force_utf8])
),
ok.
%% @doc This function is exported for scripts/schema-dump-reformat.escript
reformat_schema_dump(StructsJsonArray0) ->
%% prepare
StructsJsonArray = deduplicate_by_full_name(StructsJsonArray0),
#{fields := RootFields} = hd(StructsJsonArray),
RootNames0 = lists:map(fun(#{name := RootName}) -> RootName end, RootFields),
RootNames = lists:map(fun to_bin/1, RootNames0),
%% reformat
[Root | FlatStructs0] = lists:map(
fun(Struct) -> gen_flat_doc(RootNames, Struct) end, StructsJsonArray
),
FlatStructs = [Root#{text => <<"root">>, hash => <<"root">>} | FlatStructs0],
gen_nested_doc(FlatStructs).
deduplicate_by_full_name(Structs) ->
deduplicate_by_full_name(Structs, #{}, []).
deduplicate_by_full_name([], _Seen, Acc) ->
lists:reverse(Acc);
deduplicate_by_full_name([#{full_name := FullName} = H | T], Seen, Acc) ->
case maps:get(FullName, Seen, false) of
false ->
deduplicate_by_full_name(T, Seen#{FullName => H}, [H | Acc]);
H ->
%% Name clash, but identical, ignore
deduplicate_by_full_name(T, Seen, Acc);
_Different ->
%% ADD NAMESPACE!
throw({duplicate_full_name, FullName})
end.
%% Ggenerate nested docs from root struct.
%% Due to the fact that the same struct can be referenced by multiple fields,
%% we need to generate a unique nested doc for each reference.
%% The unique path to each type and is of the below format:
%% - A a path starts either with 'T-' or 'V-'. T stands for type, V stands for value.
%% - A path is a list of strings delimited by '-'.
%% - The letter S is used to separate struct name from field name.
%% - Field names are however NOT denoted by a leading 'F-'.
%% For example:
%% - T-root: the root struct;
%% - T-foo-S-footype: the struct named "footype" in the foo field of root struct;
%% - V-foo-S-footype-bar: the field named "bar" in the struct named "footype" in the foo field of root struct
gen_nested_doc(Structs) ->
KeyByFullName = lists:foldl(
fun(#{hash := FullName} = Struct, Acc) ->
maps:put(FullName, Struct, Acc)
end,
#{},
Structs
),
FindFn = fun(Hash) -> maps:get(Hash, KeyByFullName) end,
gen_nested_doc(hd(Structs), FindFn, []).
gen_nested_doc(#{fields := Fields} = Struct, FindFn, Path) ->
TypeAnchor = make_type_anchor(Path),
ValueAnchor = fun(FieldName) -> make_value_anchor(Path, FieldName) end,
NewFields = lists:map(
fun(#{text := Name} = Field) ->
NewField = expand_field(Field, FindFn, Path),
NewField#{hash => ValueAnchor(Name)}
end,
Fields
),
Struct#{
fields => NewFields,
hash => TypeAnchor
}.
%% Make anchor for type.
%% Start with "T-" to distinguish from value anchor.
make_type_anchor([]) ->
<<"T-root">>;
make_type_anchor(Path) ->
to_bin(["T-", lists:join("-", lists:reverse(Path))]).
%% Value anchor is used to link to the field's struct.
%% Start with "V-" to distinguish from type anchor.
make_value_anchor(Path, FieldName) ->
to_bin(["V-", join_path_hash(Path, FieldName)]).
%% Make a globally unique "hash" (the http anchor) for each struct field.
join_path_hash([], Name) ->
Name;
join_path_hash(Path, Name) ->
to_bin(lists:join("-", lists:reverse([Name | Path]))).
%% Expand field's struct reference to nested doc.
expand_field(#{text := Name, refs := References} = Field, FindFn, Path) ->
%% Add struct type name in path to make it unique.
NewReferences = lists:map(
fun(#{text := StructName} = Ref) ->
expand_ref(Ref, FindFn, [StructName, "S", Name | Path])
end,
References
),
Field#{refs => NewReferences};
expand_field(Field, _FindFn, _Path) ->
%% No reference, no need to expand.
Field.
expand_ref(#{hash := FullName}, FindFn, Path) ->
Struct = FindFn(FullName),
gen_nested_doc(Struct, FindFn, Path).
%% generate flat docs for each struct.
%% using references to link to other structs.
gen_flat_doc(RootNames, #{full_name := FullName, fields := Fields} = S) ->
ShortName = short_name(FullName),
case is_missing_namespace(ShortName, to_bin(FullName), RootNames) of
true ->
io:format(standard_error, "WARN: no_namespace_for: ~s~n", [FullName]);
false ->
ok
end,
#{
text => short_name(FullName),
hash => format_hash(FullName),
doc => maps:get(desc, S, <<"">>),
fields => format_fields(Fields)
}.
format_fields([]) ->
[];
format_fields([Field | Fields]) ->
[format_field(Field) | format_fields(Fields)].
format_field(#{name := Name, aliases := Aliases, type := Type} = F) ->
L = [
{text, Name},
{type, format_type(Type)},
{refs, format_refs(Type)},
{aliases,
case Aliases of
[] -> undefined;
_ -> Aliases
end},
{default, maps:get(hocon, maps:get(default, F, #{}), undefined)},
{doc, maps:get(desc, F, undefined)}
],
maps:from_list([{K, V} || {K, V} <- L, V =/= undefined]).
format_refs(Type) ->
References = find_refs(Type),
case lists:map(fun format_ref/1, References) of
[] -> undefined;
L -> L
end.
format_ref(FullName) ->
#{text => short_name(FullName), hash => format_hash(FullName)}.
find_refs(Type) ->
lists:reverse(find_refs(Type, [])).
%% go deep into union, array, and map to find references
find_refs(#{kind := union, members := Members}, Acc) ->
lists:foldl(fun find_refs/2, Acc, Members);
find_refs(#{kind := array, elements := Elements}, Acc) ->
find_refs(Elements, Acc);
find_refs(#{kind := map, values := Values}, Acc) ->
find_refs(Values, Acc);
find_refs(#{kind := struct, name := FullName}, Acc) ->
[FullName | Acc];
find_refs(_, Acc) ->
Acc.
format_type(#{kind := primitive, name := Name}) ->
format_primitive_type(Name);
format_type(#{kind := singleton, name := Name}) ->
to_bin(["String(\"", to_bin(Name), "\")"]);
format_type(#{kind := enum, symbols := Symbols}) ->
CommaSep = lists:join(",", lists:map(fun(S) -> to_bin(S) end, Symbols)),
to_bin(["Enum(", CommaSep, ")"]);
format_type(#{kind := array, elements := ElementsType}) ->
to_bin(["Array(", format_type(ElementsType), ")"]);
format_type(#{kind := union, members := MemberTypes} = U) ->
DN = maps:get(display_name, U, undefined),
case DN of
undefined ->
to_bin(["OneOf(", format_union_members(MemberTypes), ")"]);
Name ->
format_primitive_type(Name)
end;
format_type(#{kind := struct, name := FullName}) ->
to_bin(["Struct(", short_name(FullName), ")"]);
format_type(#{kind := map, name := Name, values := ValuesType}) ->
to_bin(["Map($", Name, "->", format_type(ValuesType), ")"]).
format_union_members(Members) ->
format_union_members(Members, []).
format_union_members([], Acc) ->
lists:join(",", lists:reverse(Acc));
format_union_members([Member | Members], Acc) ->
NewAcc = [format_type(Member) | Acc],
format_union_members(Members, NewAcc).
format_primitive_type(TypeStr) ->
Spec = emqx_conf_schema_types:readable_docgen(?MODULE, TypeStr),
to_bin(maps:get(type, Spec)).
%% All types should have a namespace to avlid name clashing.
is_missing_namespace(ShortName, FullName, RootNames) ->
case lists:member(ShortName, RootNames) of
true ->
false;
false ->
ShortName =:= FullName
end.
%% Returns short name from full name, fullname delemited by colon(:).
short_name(FullName) ->
case string:split(FullName, ":") of
[_, Name] -> to_bin(Name);
_ -> to_bin(FullName)
end.
%% Returns the hash-anchor from full name, fullname delemited by colon(:).
format_hash(FullName) ->
case string:split(FullName, ":") of
[Namespace, Name] ->
ok = warn_bad_namespace(Namespace),
iolist_to_binary([Namespace, "__", Name]);
_ ->
iolist_to_binary(FullName)
end.
%% namespace should only have letters, numbers, and underscores.
warn_bad_namespace(Namespace) ->
case re:run(Namespace, "^[a-zA-Z0-9_]+$", [{capture, none}]) of
nomatch ->
case erlang:get({bad_namespace, Namespace}) of
true ->
ok;
_ ->
erlang:put({bad_namespace, Namespace}, true),
io:format(standard_error, "WARN: bad_namespace: ~s~n", [Namespace])
end;
_ ->
ok
end.
%% TODO: move this function to emqx_dashboard when we stop generating this JSON at build time.
hotconf_schema_json() ->
@ -306,12 +560,7 @@ hocon_schema_to_spec(Atom, _LocalModule) when is_atom(Atom) ->
typename_to_spec(TypeStr, Module) ->
emqx_conf_schema_types:readable_dashboard(Module, TypeStr).
to_bin(List) when is_list(List) ->
case io_lib:printable_list(List) of
true -> unicode:characters_to_binary(List);
false -> List
end;
to_bin(List) when is_list(List) -> iolist_to_binary(List);
to_bin(Boolean) when is_boolean(Boolean) -> Boolean;
to_bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8);
to_bin(X) ->
X.
to_bin(X) -> X.

3
apps/emqx_conf/src/emqx_conf_schema.erl
View File

@ -79,8 +79,7 @@
upgrade_raw_conf(RawConf) ->
emqx_connector_schema:transform_bridges_v1_to_connectors_and_bridges_v2(RawConf).
%% root config should not have a namespace
namespace() -> undefined.
namespace() -> emqx.
tags() ->
[<<"EMQX">>].

3
apps/emqx_dashboard_sso/src/emqx_dashboard_sso_saml.erl
View File

@ -12,6 +12,7 @@
-behaviour(emqx_dashboard_sso).
-export([
namespace/0,
hocon_ref/0,
login_ref/0,
fields/1,
@ -43,6 +44,8 @@
%% Hocon Schema
%%------------------------------------------------------------------------------
namespace() -> "dashboard".
hocon_ref() ->
hoconsc:ref(?MODULE, saml).

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

@ -21,7 +21,7 @@
%%------------------------------------------------------------------------------
%% Hocon Schema
%%------------------------------------------------------------------------------
namespace() -> "sso".
namespace() -> dashboard.
fields(sso) ->
lists:map(

4
apps/emqx_gateway_coap/src/emqx_coap_schema.erl
View File

@ -26,7 +26,9 @@
-reflect_type([duration/0]).
%% config schema provides
-export([fields/1, desc/1]).
-export([namespace/0, fields/1, desc/1]).
namespace() -> "gateway".
fields(coap) ->
[

4
apps/emqx_gateway_exproto/src/emqx_exproto_schema.erl
View File

@ -28,7 +28,9 @@
]).
%% config schema provides
-export([fields/1, desc/1]).
-export([namespace/0, fields/1, desc/1]).
namespace() -> "gateway".
fields(exproto) ->
[

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

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_gateway_exproto, [
{description, "ExProto Gateway"},
{vsn, "0.1.4"},
{vsn, "0.1.5"},
{registered, []},
{applications, [kernel, stdlib, grpc, emqx, emqx_gateway]},
{env, []},

4
apps/emqx_gateway_lwm2m/src/emqx_lwm2m_schema.erl
View File

@ -28,7 +28,9 @@
-reflect_type([duration/0, duration_s/0]).
%% config schema provides
-export([fields/1, desc/1]).
-export([namespace/0, fields/1, desc/1]).
namespace() -> gateway.
fields(lwm2m) ->
[

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

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_gateway_mqttsn, [
{description, "MQTT-SN Gateway"},
{vsn, "0.1.5"},
{vsn, "0.1.6"},
{registered, []},
{applications, [kernel, stdlib, emqx, emqx_gateway]},
{env, []},

4
apps/emqx_gateway_mqttsn/src/emqx_mqttsn_schema.erl
View File

@ -21,7 +21,9 @@
-include_lib("typerefl/include/types.hrl").
%% config schema provides
-export([fields/1, desc/1]).
-export([namespace/0, fields/1, desc/1]).
namespace() -> "gateway".
fields(mqttsn) ->
[

4
apps/emqx_gateway_stomp/src/emqx_stomp_schema.erl
View File

@ -20,7 +20,9 @@
-include_lib("typerefl/include/types.hrl").
%% config schema provides
-export([fields/1, desc/1]).
-export([namespace/0, fields/1, desc/1]).
namespace() -> "gateway".
fields(stomp) ->
[

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

@ -2,7 +2,7 @@
{application, emqx_psk, [
{description, "EMQX PSK"},
% strict semver, bump manually!
{vsn, "5.0.4"},
{vsn, "5.0.5"},
{modules, []},
{registered, [emqx_psk_sup]},
{applications, [kernel, stdlib]},

2
apps/emqx_psk/src/emqx_psk_schema.erl
View File

@ -28,7 +28,7 @@
fields/1
]).
namespace() -> "authn-psk".
namespace() -> "psk".
roots() -> ["psk_authentication"].

2
apps/emqx_retainer/src/emqx_retainer_schema.erl
View File

@ -30,7 +30,7 @@
-define(INVALID_SPEC(_REASON_), throw({_REASON_, #{default => ?DEFAULT_INDICES}})).
namespace() -> "retainer".
namespace() -> retainer.
roots() ->
[

2
apps/emqx_retainer/test/emqx_retainer_SUITE.erl
View File

@ -555,7 +555,7 @@ t_page_read(_) ->
ok = emqtt:disconnect(C1).
t_only_for_coverage(_) ->
?assertEqual("retainer", emqx_retainer_schema:namespace()),
?assertEqual(retainer, emqx_retainer_schema:namespace()),
ignored = gen_server:call(emqx_retainer, unexpected),
ok = gen_server:cast(emqx_retainer, unexpected),
unexpected = erlang:send(erlang:whereis(emqx_retainer), unexpected),

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

@ -1,6 +1,6 @@
{application, emqx_schema_registry, [
{description, "EMQX Schema Registry"},
{vsn, "0.1.7"},
{vsn, "0.1.8"},
{registered, [emqx_schema_registry_sup]},
{mod, {emqx_schema_registry_app, []}},
{included_applications, [

3
apps/emqx_schema_registry/src/emqx_schema_registry_schema.erl
View File

@ -10,6 +10,7 @@
%% `hocon_schema' API
-export([
namespace/0,
roots/0,
fields/1,
desc/1,
@ -26,6 +27,8 @@
%% `hocon_schema' APIs
%%------------------------------------------------------------------------------
namespace() -> ?CONF_KEY_ROOT.
roots() ->
[{?CONF_KEY_ROOT, mk(ref(?CONF_KEY_ROOT), #{required => false})}].

132
scripts/schema-dump-reformat.escript Executable file
View File

@ -0,0 +1,132 @@
#!/usr/bin/env escript
%% This script translates the hocon_schema_json's schema dump to a new format.
%% It is used to convert older version EMQX's schema dumps to the new format
%% after all files are upgraded to the new format, this script can be removed.
-mode(compile).
main([Input]) ->
ok = add_libs(),
_ = atoms(),
{ok, Data} = file:read_file(Input),
Json = jsx:decode(Data),
NewJson = reformat(Json),
io:format("~s~n", [jsx:encode(NewJson)]);
main(_) ->
io:format("Usage: schema-dump-reformat.escript <input.json>~n"),
halt(1).
reformat(Json) ->
emqx_conf:reformat_schema_dump(fix(Json)).
%% fix old type specs to make them compatible with new type specs
fix(#{
<<"kind">> := <<"union">>,
<<"members">> := [#{<<"name">> := <<"string()">>}, #{<<"name">> := <<"function()">>}]
}) ->
%% s3_exporter.secret_access_key
#{
kind => primitive,
name => <<"string()">>
};
fix(#{<<"kind">> := <<"primitive">>, <<"name">> := <<"emqx_conf_schema:log_level()">>}) ->
#{
kind => enum,
symbols => [emergency, alert, critical, error, warning, notice, info, debug, none, all]
};
fix(#{<<"kind">> := <<"primitive">>, <<"name">> := <<"emqx_connector_http:pool_type()">>}) ->
#{kind => enum, symbols => [random, hash]};
fix(#{<<"kind">> := <<"primitive">>, <<"name">> := <<"emqx_bridge_http_connector:pool_type()">>}) ->
#{kind => enum, symbols => [random, hash]};
fix(Map) when is_map(Map) ->
maps:from_list(fix(maps:to_list(Map)));
fix(List) when is_list(List) ->
lists:map(fun fix/1, List);
fix({<<"kind">>, Kind}) ->
{kind, binary_to_atom(Kind, utf8)};
fix({<<"name">>, Type}) ->
{name, fix_type(Type)};
fix({K, V}) ->
{binary_to_atom(K, utf8), fix(V)};
fix(V) when is_number(V) ->
V;
fix(V) when is_atom(V) ->
V;
fix(V) when is_binary(V) ->
V.
%% ensure below ebin dirs are added to code path:
%% _build/default/lib/*/ebin
%% _build/emqx/lib/*/ebin
%% _build/emqx-enterprise/lib/*/ebin
add_libs() ->
Profile = os:getenv("PROFILE"),
case Profile of
"emqx" ->
ok;
"emqx-enterprise" ->
ok;
_ ->
io:format("PROFILE is not set~n"),
halt(1)
end,
Dirs =
filelib:wildcard("_build/default/lib/*/ebin") ++
filelib:wildcard("_build/" ++ Profile ++ "/lib/*/ebin"),
lists:foreach(fun add_lib/1, Dirs).
add_lib(Dir) ->
code:add_patha(Dir),
Beams = filelib:wildcard(Dir ++ "/*.beam"),
_ = spawn(fun() -> lists:foreach(fun load_beam/1, Beams) end),
ok.
load_beam(BeamFile) ->
ModuleName = filename:basename(BeamFile, ".beam"),
Module = list_to_atom(ModuleName),
%% load the beams to make sure the atoms are existing
code:ensure_loaded(Module),
ok.
fix_type(<<"[{string(), string()}]">>) ->
<<"map()">>;
fix_type(<<"[{binary(), binary()}]">>) ->
<<"map()">>;
fix_type(<<"emqx_limiter_schema:rate()">>) ->
<<"string()">>;
fix_type(<<"emqx_limiter_schema:burst_rate()">>) ->
<<"string()">>;
fix_type(<<"emqx_limiter_schema:capacity()">>) ->
<<"string()">>;
fix_type(<<"emqx_limiter_schema:initial()">>) ->
<<"string()">>;
fix_type(<<"emqx_limiter_schema:failure_strategy()">>) ->
<<"string()">>;
fix_type(<<"emqx_conf_schema:file()">>) ->
<<"string()">>;
fix_type(<<"#{term() => binary()}">>) ->
<<"map()">>;
fix_type(<<"[term()]">>) ->
%% jwt claims
<<"map()">>;
fix_type(<<"emqx_ee_bridge_influxdb:write_syntax()">>) ->
<<"string()">>;
fix_type(<<"emqx_bridge_influxdb:write_syntax()">>) ->
<<"string()">>;
fix_type(<<"emqx_schema:mqtt_max_packet_size()">>) ->
<<"non_neg_integer()">>;
fix_type(<<"emqx_s3_schema:secret_access_key()">>) ->
<<"string()">>;
fix_type(Type) ->
Type.
%% ensure atoms are loaded
%% these atoms are from older version of emqx
atoms() ->
[
emqx_ee_connector_clickhouse,
emqx_ee_bridge_gcp_pubsub,
emqx_ee_bridge_influxdb,
emqx_connector_http
].