feat(ft-api): provide configuration API

To configure `emqx_ft` during the runtime.

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -118,7 +118,7 @@
 -type route_path() :: string() | binary().
 -type route_methods() :: map().
 -type route_handler() :: atom().
--type route_options() :: #{filter => filter() | undefined}.
+-type route_options() :: #{filter => filter()}.
 
 -type api_spec_entry() :: {route_path(), route_methods(), route_handler(), route_options()}.
 -type api_spec_component() :: map().
@@ -137,10 +137,9 @@ spec(Module, Options) ->
     {ApiSpec, AllRefs} =
         lists:foldl(
             fun(Path, {AllAcc, AllRefsAcc}) ->
-                {OperationId, Specs, Refs} = parse_spec_ref(Module, Path, Options),
-                Opts = #{filter => filter(Options)},
+                {OperationId, Specs, Refs, RouteOpts} = parse_spec_ref(Module, Path, Options),
                 {
-                    [{filename:join("/", Path), Specs, OperationId, Opts} | AllAcc],
+                    [{filename:join("/", Path), Specs, OperationId, RouteOpts} | AllAcc],
                     Refs ++ AllRefsAcc
                 }
             end,
@@ -350,6 +349,7 @@ parse_spec_ref(Module, Path, Options) ->
                 ),
                 error({failed_to_generate_swagger_spec, Module, Path})
         end,
+    OperationId = maps:get('operationId', Schema),
     {Specs, Refs} = maps:fold(
         fun(Method, Meta, {Acc, RefsAcc}) ->
             (not lists:member(Method, ?METHODS)) andalso
@@ -358,9 +358,13 @@ parse_spec_ref(Module, Path, Options) ->
             {Acc#{Method => Spec}, SubRefs ++ RefsAcc}
         end,
         {#{}, []},
-        maps:without(['operationId'], Schema)
+        maps:without(['operationId', 'filter'], Schema)
     ),
-    {maps:get('operationId', Schema), Specs, Refs}.
+    RouteOpts = generate_route_opts(Schema, Options),
+    {OperationId, Specs, Refs, RouteOpts}.
+
+generate_route_opts(Schema, Options) ->
+    #{filter => compose_filters(filter(Options), custom_filter(Schema))}.
 
 check_parameters(Request, Spec, Module) ->
     #{bindings := Bindings, query_string := QueryStr} = Request,

apps/emqx_dashboard/test/emqx_swagger_parameter_SUITE.erl

@@ -108,8 +108,12 @@ t_ref(_Config) ->
     LocalPath = "/test/in/ref/local",
     Path = "/test/in/ref",
     Expect = [#{<<"$ref">> => <<"#/components/parameters/emqx_swagger_parameter_SUITE.page">>}],
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, LocalPath, #{}),
+    {OperationId, Spec, Refs, RouteOpts} = emqx_dashboard_swagger:parse_spec_ref(
+        ?MODULE, Path, #{}
+    ),
+    {OperationId, Spec, Refs, RouteOpts} = emqx_dashboard_swagger:parse_spec_ref(
+        ?MODULE, LocalPath, #{}
+    ),
     ?assertEqual(test, OperationId),
     Params = maps:get(parameters, maps:get(post, Spec)),
     ?assertEqual(Expect, Params),
@@ -122,7 +126,7 @@ t_public_ref(_Config) ->
         #{<<"$ref">> => <<"#/components/parameters/public.page">>},
         #{<<"$ref">> => <<"#/components/parameters/public.limit">>}
     ],
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     Params = maps:get(parameters, maps:get(post, Spec)),
     ?assertEqual(Expect, Params),
@@ -264,7 +268,7 @@ t_nullable(_Config) ->
 t_method(_Config) ->
     PathOk = "/method/ok",
     PathError = "/method/error",
-    {test, Spec, []} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, PathOk, #{}),
+    {test, Spec, [], #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, PathOk, #{}),
     ?assertEqual(lists:sort(?METHODS), lists:sort(maps:keys(Spec))),
     ?assertThrow(
         {error, #{module := ?MODULE, path := PathError, method := bar}},
@@ -393,7 +397,7 @@ assert_all_filters_equal(Spec, Filter) ->
     ).
 
 validate(Path, ExpectParams) ->
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     Params = maps:get(parameters, maps:get(post, Spec)),
     ?assertEqual(ExpectParams, Params),

apps/emqx_dashboard/test/emqx_swagger_requestBody_SUITE.erl

@@ -719,7 +719,7 @@ t_object_trans_error(_Config) ->
     ok.
 
 validate(Path, ExpectSpec, ExpectRefs) ->
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     ?assertEqual(ExpectSpec, Spec),
     ?assertEqual(ExpectRefs, Refs),

apps/emqx_dashboard/test/emqx_swagger_response_SUITE.erl

@@ -129,7 +129,7 @@ t_error(_Config) ->
                 }
             }
     },
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     Response = maps:get(responses, maps:get(get, Spec)),
     ?assertEqual(Error400, maps:get(<<"400">>, Response)),
@@ -375,7 +375,7 @@ t_complicated_type(_Config) ->
                 }
         }
     },
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     Response = maps:get(responses, maps:get(post, Spec)),
     ?assertEqual(Object, maps:get(<<"200">>, Response)),
@@ -665,7 +665,7 @@ schema("/fields/sub") ->
     to_schema(hoconsc:ref(sub_fields)).
 
 validate(Path, ExpectObject, ExpectRefs) ->
-    {OperationId, Spec, Refs} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
+    {OperationId, Spec, Refs, #{}} = emqx_dashboard_swagger:parse_spec_ref(?MODULE, Path, #{}),
     ?assertEqual(test, OperationId),
     Response = maps:get(responses, maps:get(post, Spec)),
     ?assertEqual(ExpectObject, maps:get(<<"200">>, Response)),

apps/emqx_ft/src/emqx_ft_api.erl

@@ -40,27 +40,30 @@
 %% API callbacks
 -export([
     '/file_transfer/files'/2,
-    '/file_transfer/files/:clientid/:fileid'/2
+    '/file_transfer/files/:clientid/:fileid'/2,
+    '/file_transfer'/2
 ]).
 
 -import(hoconsc, [mk/2, ref/1, ref/2]).
 
+-define(SCHEMA_CONFIG, ref(emqx_ft_schema, file_transfer)).
+
 namespace() -> "file_transfer".
 
 api_spec() ->
-    emqx_dashboard_swagger:spec(?MODULE, #{
-        check_schema => true, filter => fun ?MODULE:check_ft_enabled/2
-    }).
+    emqx_dashboard_swagger:spec(?MODULE, #{check_schema => true}).
 
 paths() ->
     [
         "/file_transfer/files",
-        "/file_transfer/files/:clientid/:fileid"
+        "/file_transfer/files/:clientid/:fileid",
+        "/file_transfer"
     ].
 
 schema("/file_transfer/files") ->
     #{
         'operationId' => '/file_transfer/files',
+        filter => fun ?MODULE:check_ft_enabled/2,
         get => #{
             tags => ?TAGS,
             summary => <<"List all uploaded files">>,
@@ -83,6 +86,7 @@ schema("/file_transfer/files") ->
 schema("/file_transfer/files/:clientid/:fileid") ->
     #{
         'operationId' => '/file_transfer/files/:clientid/:fileid',
+        filter => fun ?MODULE:check_ft_enabled/2,
         get => #{
             tags => ?TAGS,
             summary => <<"List files uploaded in a specific transfer">>,
@@ -101,6 +105,36 @@ schema("/file_transfer/files/:clientid/:fileid") ->
                 )
             }
         }
+    };
+schema("/file_transfer") ->
+    #{
+        'operationId' => '/file_transfer',
+        get => #{
+            tags => [<<"file_transfer">>],
+            summary => <<"Get current File Transfer configuration">>,
+            description => ?DESC("file_transfer_get_config"),
+            responses => #{
+                200 => ?SCHEMA_CONFIG,
+                503 => emqx_dashboard_swagger:error_codes(
+                    ['SERVICE_UNAVAILABLE'], error_desc('SERVICE_UNAVAILABLE')
+                )
+            }
+        },
+        put => #{
+            tags => [<<"file_transfer">>],
+            summary => <<"Update File Transfer configuration">>,
+            description => ?DESC("file_transfer_update_config"),
+            'requestBody' => ?SCHEMA_CONFIG,
+            responses => #{
+                200 => ?SCHEMA_CONFIG,
+                400 => emqx_dashboard_swagger:error_codes(
+                    ['INVALID_CONFIG'], error_desc('INVALID_CONFIG')
+                ),
+                503 => emqx_dashboard_swagger:error_codes(
+                    ['SERVICE_UNAVAILABLE'], error_desc('SERVICE_UNAVAILABLE')
+                )
+            }
+        }
     }.
 
 check_ft_enabled(Params, _Meta) ->
@@ -108,7 +142,7 @@ check_ft_enabled(Params, _Meta) ->
         true ->
             {ok, Params};
         false ->
-            {503, error_msg('SERVICE_UNAVAILABLE', <<"Service unavailable">>)}
+            {503, error_msg('SERVICE_UNAVAILABLE')}
     end.
 
 '/file_transfer/files'(get, #{
@@ -147,6 +181,18 @@ check_ft_enabled(Params, _Meta) ->
             {503, error_msg('SERVICE_UNAVAILABLE')}
     end.
 
+'/file_transfer'(get, _Meta) ->
+    {200, format_config(emqx_ft_conf:get())};
+'/file_transfer'(put, #{body := ConfigIn}) ->
+    case emqx_ft_conf:update(ConfigIn) of
+        {ok, #{config := Config}} ->
+            {200, format_config(Config)};
+        {error, Error = #{kind := validation_error}} ->
+            {400, error_msg('INVALID_CONFIG', format_validation_error(Error))};
+        {error, Error} ->
+            {400, error_msg('INVALID_CONFIG', emqx_utils:format(Error))}
+    end.
+
 format_page(#{items := Files, cursor := Cursor}) ->
     #{
         <<"files">> => lists:map(fun format_file_info/1, Files),
@@ -157,14 +203,23 @@ format_page(#{items := Files}) ->
         <<"files">> => lists:map(fun format_file_info/1, Files)
     }.
 
+format_config(Config) ->
+    Schema = emqx_hocon:make_schema(emqx_ft_schema:fields(file_transfer)),
+    hocon_tconf:make_serializable(Schema, emqx_utils_maps:binary_key_map(Config), #{}).
+
+format_validation_error(Error) ->
+    emqx_logger_jsonfmt:best_effort_json(Error).
+
 error_msg(Code) ->
     #{code => Code, message => error_desc(Code)}.
 
 error_msg(Code, Msg) ->
-    #{code => Code, message => emqx_utils:readable_error_msg(Msg)}.
+    #{code => Code, message => Msg}.
 
 error_desc('FILES_NOT_FOUND') ->
     <<"Files requested for this transfer could not be found">>;
+error_desc('INVALID_CONFIG') ->
+    <<"Provided configuration is invalid">>;
 error_desc('SERVICE_UNAVAILABLE') ->
     <<"Service unavailable">>.
 

apps/emqx_ft/test/emqx_ft_api_SUITE.erl

@@ -108,6 +108,11 @@ init_per_testcase(Case, Config) ->
     [{tc, Case} | Config].
 end_per_testcase(t_ft_disabled, _Config) ->
     emqx_config:put([file_transfer, enable], true);
+end_per_testcase(t_configure, Config) ->
+    {ok, 200, _} = request(put, uri(["file_transfer"]), #{
+        <<"enable">> => true,
+        <<"storage">> => emqx_ft_test_helpers:local_storage(Config)
+    });
 end_per_testcase(_Case, _Config) ->
     ok.
 
@@ -310,6 +315,155 @@ t_ft_disabled(Config) ->
         )
     ).
 
+t_configure(Config) ->
+    ?assertMatch(
+        {ok, 200, #{<<"enable">> := true, <<"storage">> := #{}}},
+        request_json(get, uri(["file_transfer"]), Config)
+    ),
+    ?assertMatch(
+        {ok, 200, #{<<"enable">> := false}},
+        request_json(put, uri(["file_transfer"]), #{<<"enable">> => false}, Config)
+    ),
+    ?assertMatch(
+        {ok, 200, #{<<"enable">> := false}},
+        request_json(get, uri(["file_transfer"]), Config)
+    ),
+    ?assertMatch(
+        {ok, 200, #{}},
+        request_json(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => emqx_ft_test_helpers:local_storage(Config)
+            },
+            Config
+        )
+    ),
+    ?assertMatch(
+        {ok, 400, _},
+        request(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => #{
+                    <<"local">> => #{},
+                    <<"remote">> => #{}
+                }
+            },
+            Config
+        )
+    ),
+    ?assertMatch(
+        {ok, 400, _},
+        request(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => #{
+                    <<"local">> => #{
+                        <<"gc">> => #{<<"interval">> => -42}
+                    }
+                }
+            },
+            Config
+        )
+    ),
+    S3Exporter = #{
+        <<"host">> => <<"localhost">>,
+        <<"port">> => 9000,
+        <<"bucket">> => <<"emqx">>,
+        <<"transport_options">> => #{
+            <<"ssl">> => #{
+                <<"enable">> => true,
+                <<"certfile">> => emqx_ft_test_helpers:pem_privkey(),
+                <<"keyfile">> => emqx_ft_test_helpers:pem_privkey()
+            }
+        }
+    },
+    ?assertMatch(
+        {ok, 200, #{
+            <<"enable">> := true,
+            <<"storage">> := #{
+                <<"local">> := #{
+                    <<"exporter">> := #{
+                        <<"s3">> := #{
+                            <<"transport_options">> := #{
+                                <<"ssl">> := #{
+                                    <<"enable">> := true,
+                                    <<"certfile">> := <<"/", _CertFilepath/bytes>>,
+                                    <<"keyfile">> := <<"/", _KeyFilepath/bytes>>
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }},
+        request_json(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => #{
+                    <<"local">> => #{
+                        <<"exporter">> => #{
+                            <<"s3">> => S3Exporter
+                        }
+                    }
+                }
+            },
+            Config
+        )
+    ),
+    ?assertMatch(
+        {ok, 400, _},
+        request_json(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => #{
+                    <<"local">> => #{
+                        <<"exporter">> => #{
+                            <<"s3">> => emqx_utils_maps:deep_put(
+                                [<<"transport_options">>, <<"ssl">>, <<"keyfile">>],
+                                S3Exporter,
+                                <<>>
+                            )
+                        }
+                    }
+                }
+            },
+            Config
+        )
+    ),
+    ?assertMatch(
+        {ok, 200, #{}},
+        request_json(
+            put,
+            uri(["file_transfer"]),
+            #{
+                <<"enable">> => true,
+                <<"storage">> => #{
+                    <<"local">> => #{
+                        <<"exporter">> => #{
+                            <<"s3">> => emqx_utils_maps:deep_put(
+                                [<<"transport_options">>, <<"ssl">>, <<"enable">>],
+                                S3Exporter,
+                                false
+                            )
+                        }
+                    }
+                }
+            },
+            Config
+        )
+    ),
+    ok.
+
 %%--------------------------------------------------------------------
 %% Helpers
 %%--------------------------------------------------------------------
@@ -332,17 +486,26 @@ mk_file_name(N) ->
     "file." ++ integer_to_list(N).
 
 request(Method, Url, Config) ->
+    request(Method, Url, [], Config).
+
+request(Method, Url, Body, Config) ->
     Opts = #{compatible_mode => true, httpc_req_opts => [{body_format, binary}]},
-    emqx_mgmt_api_test_util:request_api(Method, Url, [], auth_header(Config), [], Opts).
+    request(Method, Url, Body, Opts, Config).
 
-request_json(Method, Url, Config) ->
-    case request(Method, Url, Config) of
+request(Method, Url, Body, Opts, Config) ->
+    emqx_mgmt_api_test_util:request_api(Method, Url, Body, auth_header(Config), [], Opts).
+
+request_json(Method, Url, Body, Config) ->
+    case request(Method, Url, Body, [], Config) of
         {ok, Code, Body} ->
             {ok, Code, json(Body)};
         Otherwise ->
             Otherwise
     end.
 
+request_json(Method, Url, Config) ->
+    request_json(Method, Url, [], Config).
+
 json(Body) when is_binary(Body) ->
     emqx_utils_json:decode(Body, [return_maps]).
 

apps/emqx_utils/src/emqx_utils.erl

@@ -60,7 +60,8 @@
     safe_filename/1,
     diff_lists/3,
     merge_lists/3,
-    tcp_keepalive_opts/4
+    tcp_keepalive_opts/4,
+    format/1
 ]).
 
 -export([
@@ -525,6 +526,9 @@ tcp_keepalive_opts({unix, darwin}, Idle, Interval, Probes) ->
 tcp_keepalive_opts(OS, _Idle, _Interval, _Probes) ->
     {error, {unsupported_os, OS}}.
 
+format(Term) ->
+    iolist_to_binary(io_lib:format("~0p", [Term])).
+
 %%------------------------------------------------------------------------------
 %% Internal Functions
 %%------------------------------------------------------------------------------
@@ -606,7 +610,7 @@ to_hr_error({not_authorized, _}) ->
 to_hr_error({malformed_username_or_password, _}) ->
     <<"Bad username or password">>;
 to_hr_error(Error) ->
-    iolist_to_binary(io_lib:format("~0p", [Error])).
+    format(Error).
 
 try_to_existing_atom(Convert, Data, Encoding) ->
     try Convert(Data, Encoding) of

rel/i18n/emqx_ft_api.hocon

@@ -6,6 +6,12 @@ file_list.desc:
 file_list_transfer.desc:
 """List a file uploaded during specified transfer, identified by client id and file id."""
 
+file_transfer_get_config.desc:
+"""Show current File Transfer configuration."""
+
+file_transfer_update_config.desc:
+"""Replace File Transfer configuration."""
+
 }
 
 emqx_ft_storage_exporter_fs_api {