Merge pull request #6809 from k32/bpapi-resource

refactor(emqx_resource): Decorate RPCs

apps/emqx/test/emqx_bpapi_static_checks.erl

@@ -49,13 +49,17 @@
 %% List of known RPC backend modules:
 -define(RPC_MODULES, "gen_rpc, erpc, rpc, emqx_rpc").
 %% List of known functions also known to do RPC:
--define(RPC_FUNCTIONS, "emqx_cluster_rpc:multicall/3, emqx_cluster_rpc:multicall/5").
+-define(RPC_FUNCTIONS, "emqx_cluster_rpc:multicall/3, emqx_cluster_rpc:multicall/5, "
+                       "emqx_plugin_libs_rule:cluster_call/3").
 %% List of functions in the RPC backend modules that we can ignore:
--define(IGNORED_RPC_CALLS, "gen_rpc:nodes/0, emqx_rpc:unwrap_erpc/1").
+-define(IGNORED_RPC_CALLS, "gen_rpc:nodes/0, emqx_rpc:unwrap_erpc/1, rpc:pmap/3"). % TODO: handle pmap
 %% List of business-layer functions that are exempt from the checks:
--define(EXEMPTIONS, "emqx_mgmt_api:do_query/6"  % Reason: legacy code. A fun and a QC query are
+-define(EXEMPTIONS,
+        "emqx_mgmt_api:do_query/6,"             % Reason: legacy code. A fun and a QC query are
                                                 % passed in the args, it's futile to try to statically
                                                 % check it
+        "emqx_plugin_libs_rule:cluster_call/3"  % Reason: some sort of external plugin API that we
+                                                % don't want to break?
        ).
 
 -define(XREF, myxref).

apps/emqx_conf/src/emqx_cluster_rpc.erl

@@ -29,7 +29,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          handle_continue/2, code_change/3]).
 
--export_type([txn_id/0, succeed_num/0, multicall_return/0]).
+-export_type([txn_id/0, succeed_num/0, multicall_return/1, multicall_return/0]).
 
 -ifdef(TEST).
 -compile(export_all).
@@ -48,9 +48,11 @@
 
 -type succeed_num() :: pos_integer() | all.
 
--type multicall_return() :: {ok, txn_id(), _Result}
-                          | {error, term()}
-                          | {retry, txn_id(), _Result, node()}.
+-type multicall_return(Result) :: {ok, txn_id(), Result}
+                                | {error, term()}
+                                | {retry, txn_id(), Result, node()}.
+
+-type multicall_return() :: multicall_return(_).
 
 %%%===================================================================
 %%% API

apps/emqx_resource/rebar.config

@@ -15,4 +15,5 @@
            ]}.
 
 {deps, [ {jsx, {git, "https://github.com/talentdeficit/jsx", {tag, "v3.1.0"}}}
+       , {emqx, {path, "../emqx"}}
        ]}.

apps/emqx_resource/src/emqx_resource.erl

@@ -146,7 +146,7 @@ create(InstId, ResourceType, Config) ->
 -spec create(instance_id(), resource_type(), resource_config(), create_opts()) ->
     {ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
 create(InstId, ResourceType, Config, Opts) ->
-    cluster_call(create_local, [InstId, ResourceType, Config, Opts]).
+    wrap_rpc(emqx_resource_proto_v1:create(InstId, ResourceType, Config, Opts)).
 
 -spec create_local(instance_id(), resource_type(), resource_config()) ->
     {ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
@@ -161,7 +161,7 @@ create_local(InstId, ResourceType, Config, Opts) ->
 -spec create_dry_run(resource_type(), resource_config()) ->
     ok | {error, Reason :: term()}.
 create_dry_run(ResourceType, Config) ->
-    cluster_call(create_dry_run_local, [ResourceType, Config]).
+    wrap_rpc(emqx_resource_proto_v1:create_dry_run(ResourceType, Config)).
 
 -spec create_dry_run_local(resource_type(), resource_config()) ->
     ok | {error, Reason :: term()}.
@@ -171,7 +171,7 @@ create_dry_run_local(ResourceType, Config) ->
 -spec recreate(instance_id(), resource_type(), resource_config(), create_opts()) ->
     {ok, resource_data()} | {error, Reason :: term()}.
 recreate(InstId, ResourceType, Config, Opts) ->
-    cluster_call(recreate_local, [InstId, ResourceType, Config, Opts]).
+    wrap_rpc(emqx_resource_proto_v1:recreate(InstId, ResourceType, Config, Opts)).
 
 -spec recreate_local(instance_id(), resource_type(), resource_config(), create_opts()) ->
     {ok, resource_data()} | {error, Reason :: term()}.
@@ -180,7 +180,7 @@ recreate_local(InstId, ResourceType, Config, Opts) ->
 
 -spec remove(instance_id()) -> ok | {error, Reason :: term()}.
 remove(InstId) ->
-    cluster_call(remove_local, [InstId]).
+    wrap_rpc(emqx_resource_proto_v1:remove(InstId)).
 
 -spec remove_local(instance_id()) -> ok | {error, Reason :: term()}.
 remove_local(InstId) ->
@@ -362,8 +362,8 @@ call_instance(InstId, Query) ->
 safe_apply(Func, Args) ->
     ?SAFE_CALL(erlang:apply(Func, Args)).
 
-cluster_call(Func, Args) ->
-    case emqx_cluster_rpc:multicall(?MODULE, Func, Args) of
+wrap_rpc(Ret) ->
+    case Ret of
         {ok, _TxnId, Result} -> Result;
         Failed -> Failed
     end.

apps/emqx_resource/src/proto/emqx_resource_proto_v1.erl

@@ -0,0 +1,62 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_resource_proto_v1).
+
+-behaviour(emqx_bpapi).
+
+-export([ introduced_in/0
+
+        , create/4
+        , create_dry_run/2
+        , recreate/4
+        , remove/1
+        ]).
+
+-include_lib("emqx/include/bpapi.hrl").
+
+introduced_in() ->
+    "5.0.0".
+
+-spec create( emqx_resource:instance_id()
+            , emqx_resource:resource_type()
+            , emqx_resource:resource_config()
+            , emqx_resource:create_opts()
+            ) ->
+          emqx_cluster_rpc:multicall_return(emqx_resource:resource_data()).
+create(InstId, ResourceType, Config, Opts) ->
+    emqx_cluster_rpc:multicall(emqx_resource, create_local, [InstId, ResourceType, Config, Opts]).
+
+-spec create_dry_run( emqx_resource:resource_type()
+                    , emqx_resource:resource_config()
+                    ) ->
+          emqx_cluster_rpc:multicall_return(emqx_resource:resource_data()).
+create_dry_run(ResourceType, Config) ->
+    emqx_cluster_rpc:multicall(emqx_resource, create_dry_run_local, [ResourceType, Config]).
+
+-spec recreate( emqx_resource:instance_id()
+              , emqx_resource:resource_type()
+              , emqx_resource:resource_config()
+              , emqx_resource:create_opts()
+              ) ->
+          emqx_cluster_rpc:multicall_return(emqx_resource:resource_data()).
+recreate(InstId, ResourceType, Config, Opts) ->
+    emqx_cluster_rpc:multicall(emqx_resource, recreate_local, [InstId, ResourceType, Config, Opts]).
+
+-spec remove(emqx_resource:instance_id()) ->
+          emqx_cluster_rpc:multicall_return(ok).
+remove(InstId) ->
+    emqx_cluster_rpc:multicall(emqx_resource, remove_local, [InstId]).