test: add tracepoints

apps/emqx_bridge/test/emqx_bridge_testlib.erl

@@ -203,7 +203,7 @@ create_rule_and_action_http(BridgeType, RuleTopic, Config) ->
 %% Testcases
 %%------------------------------------------------------------------------------
 
-t_sync_query(Config, MakeMessageFun, IsSuccessCheck) ->
+t_sync_query(Config, MakeMessageFun, IsSuccessCheck, TracePoint) ->
     ResourceId = resource_id(Config),
     ?check_trace(
         begin
@@ -217,11 +217,13 @@ t_sync_query(Config, MakeMessageFun, IsSuccessCheck) ->
             IsSuccessCheck(emqx_resource:simple_sync_query(ResourceId, Message)),
             ok
         end,
-        []
+        fun(Trace) ->
+            ?assertMatch([#{instance_id := ResourceId}], ?of_kind(TracePoint, Trace))
+        end
     ),
     ok.
 
-t_async_query(Config, MakeMessageFun, IsSuccessCheck) ->
+t_async_query(Config, MakeMessageFun, IsSuccessCheck, TracePoint) ->
     ResourceId = resource_id(Config),
     ReplyFun =
         fun(Pid, Result) ->
@@ -236,10 +238,21 @@ t_async_query(Config, MakeMessageFun, IsSuccessCheck) ->
                 ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
             ),
             Message = {send_message, MakeMessageFun()},
-            emqx_resource:query(ResourceId, Message, #{async_reply_fun => {ReplyFun, [self()]}}),
+            ?assertMatch(
+                {ok, {ok, _}},
+                ?wait_async_action(
+                    emqx_resource:query(ResourceId, Message, #{
+                        async_reply_fun => {ReplyFun, [self()]}
+                    }),
+                    #{?snk_kind := TracePoint, instance_id := ResourceId},
+                    5_000
+                )
+            ),
             ok
         end,
-        []
+        fun(Trace) ->
+            ?assertMatch([#{instance_id := ResourceId}], ?of_kind(TracePoint, Trace))
+        end
     ),
     receive
         {result, Result} -> IsSuccessCheck(Result)
@@ -318,7 +331,7 @@ t_start_stop(Config, StopTracePoint) ->
         end,
         fun(Trace) ->
             %% one for each probe, one for real
-            ?assertMatch([_, _, _], ?of_kind(StopTracePoint, Trace)),
+            ?assertMatch([_, _, #{instance_id := ResourceId}], ?of_kind(StopTracePoint, Trace)),
             ok
         end
     ),

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_impl.erl

@@ -72,7 +72,7 @@ on_start(InstanceId, Config) ->
                 instance_id => InstanceId,
                 request => maps:get(request, State, <<>>)
             }),
-            ?tp(iotdb_bridge_started, #{}),
+            ?tp(iotdb_bridge_started, #{instance_id => InstanceId}),
             {ok, maps:merge(Config, State)};
         {error, Reason} ->
             ?SLOG(error, #{
@@ -104,6 +104,7 @@ on_get_status(InstanceId, State) ->
     | {ok, pos_integer(), [term()]}
     | {error, term()}.
 on_query(InstanceId, {send_message, Message}, State) ->
+    ?tp(iotdb_bridge_on_query, #{instance_id => InstanceId}),
     ?SLOG(debug, #{
         msg => "iotdb_bridge_on_query_called",
         instance_id => InstanceId,
@@ -124,6 +125,7 @@ on_query(InstanceId, {send_message, Message}, State) ->
 -spec on_query_async(manager_id(), {send_message, map()}, {function(), [term()]}, state()) ->
     {ok, pid()} | {error, empty_request}.
 on_query_async(InstanceId, {send_message, Message}, ReplyFunAndArgs0, State) ->
+    ?tp(iotdb_bridge_on_query_async, #{instance_id => InstanceId}),
     ?SLOG(debug, #{
         msg => "iotdb_bridge_on_query_async_called",
         instance_id => InstanceId,

apps/emqx_bridge_iotdb/test/emqx_bridge_iotdb_impl_SUITE.erl

@@ -175,7 +175,7 @@ t_sync_query_simple(Config) ->
         fun(Result) ->
             ?assertEqual(ok, element(1, Result))
         end,
-    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck).
+    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck, iotdb_bridge_on_query).
 
 t_async_query(Config) ->
     DeviceId = iotdb_device(Config),
@@ -185,7 +185,9 @@ t_async_query(Config) ->
         fun(Result) ->
             ?assertEqual(ok, element(1, Result))
         end,
-    emqx_bridge_testlib:t_async_query(Config, MakeMessageFun, IsSuccessCheck).
+    emqx_bridge_testlib:t_async_query(
+        Config, MakeMessageFun, IsSuccessCheck, iotdb_bridge_on_query_async
+    ).
 
 t_sync_query_aggregated(Config) ->
     DeviceId = iotdb_device(Config),
@@ -212,7 +214,7 @@ t_sync_query_aggregated(Config) ->
         fun(Result) ->
             ?assertEqual(ok, element(1, Result))
         end,
-    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck).
+    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck, iotdb_bridge_on_query).
 
 t_sync_query_fail(Config) ->
     DeviceId = iotdb_device(Config),
@@ -222,7 +224,7 @@ t_sync_query_fail(Config) ->
         fun(Result) ->
             ?assertEqual(error, element(1, Result))
         end,
-    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck).
+    emqx_bridge_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck, iotdb_bridge_on_query).
 
 t_create_via_http(Config) ->
     emqx_bridge_testlib:t_create_via_http(Config).