chore(ds-shared-sub): lower log level

apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl

@@ -241,14 +241,14 @@ schedule_subscribe(
             ScheduledActions1 = ScheduledActions0#{
                 ShareTopicFilter => ScheduledAction#{type => {?schedule_subscribe, SubOpts}}
             },
-            ?tp(warning, shared_subs_schedule_subscribe_override, #{
+            ?tp(debug, shared_subs_schedule_subscribe_override, #{
                 share_topic_filter => ShareTopicFilter,
                 new_type => {?schedule_subscribe, SubOpts},
                 old_action => format_schedule_action(ScheduledAction)
             }),
             SharedSubS0#{scheduled_actions := ScheduledActions1};
         _ ->
-            ?tp(warning, shared_subs_schedule_subscribe_new, #{
+            ?tp(debug, shared_subs_schedule_subscribe_new, #{
                 share_topic_filter => ShareTopicFilter, subopts => SubOpts
             }),
             Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_subscribe(
@@ -299,7 +299,7 @@ schedule_unsubscribe(
             ScheduledActions1 = ScheduledActions0#{
                 ShareTopicFilter => ScheduledAction1
             },
-            ?tp(warning, shared_subs_schedule_unsubscribe_override, #{
+            ?tp(debug, shared_subs_schedule_unsubscribe_override, #{
                 share_topic_filter => ShareTopicFilter,
                 new_type => ?schedule_unsubscribe,
                 old_action => format_schedule_action(ScheduledAction0)
@@ -314,7 +314,7 @@ schedule_unsubscribe(
                     progresses => []
                 }
             },
-            ?tp(warning, shared_subs_schedule_unsubscribe_new, #{
+            ?tp(debug, shared_subs_schedule_unsubscribe_new, #{
                 share_topic_filter => ShareTopicFilter,
                 stream_keys => format_stream_keys(StreamKeys)
             }),
@@ -339,7 +339,7 @@ renew_streams(S0, #{agent := Agent0, scheduled_actions := ScheduledActions} = Sh
         Agent0
     ),
     StreamLeaseEvents =/= [] andalso
-        ?tp(warning, shared_subs_new_stream_lease_events, #{
+        ?tp(debug, shared_subs_new_stream_lease_events, #{
             stream_lease_events => format_lease_events(StreamLeaseEvents)
         }),
     S1 = lists:foldl(
@@ -506,7 +506,7 @@ run_scheduled_action(
     Progresses1 = stream_progresses(S, StreamKeysToWait0 -- StreamKeysToWait1) ++ Progresses0,
     case StreamKeysToWait1 of
         [] ->
-            ?tp(warning, shared_subs_schedule_action_complete, #{
+            ?tp(debug, shared_subs_schedule_action_complete, #{
                 share_topic_filter => ShareTopicFilter,
                 progresses => format_stream_progresses(Progresses1),
                 type => Type
@@ -530,7 +530,7 @@ run_scheduled_action(
             end;
         _ ->
             Action1 = Action#{stream_keys_to_wait => StreamKeysToWait1, progresses => Progresses1},
-            ?tp(warning, shared_subs_schedule_action_continue, #{
+            ?tp(debug, shared_subs_schedule_action_continue, #{
                 share_topic_filter => ShareTopicFilter,
                 new_action => format_schedule_action(Action1)
             }),

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_agent.erl

@@ -100,7 +100,7 @@ open(TopicSubscriptions, Opts) ->
     State0 = init_state(Opts),
     State1 = lists:foldl(
         fun({ShareTopicFilter, #{}}, State) ->
-            ?tp(warning, ds_agent_open_subscription, #{
+            ?tp(debug, ds_agent_open_subscription, #{
                 topic_filter => ShareTopicFilter
             }),
             add_shared_subscription(State, ShareTopicFilter)
@@ -120,7 +120,7 @@ can_subscribe(_State, _ShareTopicFilter, _SubOpts) ->
 
 -spec on_subscribe(t(), share_topic_filter(), emqx_types:subopts()) -> t().
 on_subscribe(State0, ShareTopicFilter, _SubOpts) ->
-    ?tp(warning, ds_agent_on_subscribe, #{
+    ?tp(debug, ds_agent_on_subscribe, #{
         share_topic_filter => ShareTopicFilter
     }),
     add_shared_subscription(State0, ShareTopicFilter).

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_group_sm.erl

@@ -133,7 +133,7 @@ new(#{
         agent => Agent,
         send_after => SendAfter
     },
-    ?tp(warning, group_sm_new, #{
+    ?tp(debug, group_sm_new, #{
         agent => Agent,
         share_topic_filter => ShareTopicFilter
     }),
@@ -176,7 +176,7 @@ handle_disconnect(
 %% Connecting state
 
 handle_connecting(#{agent := Agent, share_topic_filter := ShareTopicFilter} = GSM) ->
-    ?tp(warning, group_sm_enter_connecting, #{
+    ?tp(debug, group_sm_enter_connecting, #{
         agent => Agent,
         share_topic_filter => ShareTopicFilter
     }),
@@ -264,7 +264,7 @@ handle_leader_update_streams(
     VersionNew,
     StreamProgresses
 ) ->
-    ?tp(warning, shared_sub_group_sm_leader_update_streams, #{
+    ?tp(debug, shared_sub_group_sm_leader_update_streams, #{
         id => Id,
         version_old => VersionOld,
         version_new => VersionNew,
@@ -305,7 +305,7 @@ handle_leader_update_streams(
         maps:keys(Streams1)
     ),
     StreamLeaseEvents = AddEvents ++ RevokeEvents,
-    ?tp(warning, shared_sub_group_sm_leader_update_streams, #{
+    ?tp(debug, shared_sub_group_sm_leader_update_streams, #{
         id => Id,
         stream_lease_events => emqx_persistent_session_ds_shared_subs:format_lease_events(
             StreamLeaseEvents

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_leader.erl

@@ -132,10 +132,10 @@ init_data(#share{topic = Topic} = ShareTopicFilter, StartTime) ->
     Group = group_name(ShareTopicFilter),
     case emqx_ds_shared_sub_leader_store:open(Group) of
         Store when Store =/= false ->
-            ?tp(warning, shared_sub_leader_store_open, #{topic => ShareTopicFilter, store => Store}),
+            ?tp(debug, shared_sub_leader_store_open, #{topic => ShareTopicFilter, store => Store}),
             ok;
         false ->
-            ?tp(warning, shared_sub_leader_store_init, #{topic => ShareTopicFilter}),
+            ?tp(debug, shared_sub_leader_store_init, #{topic => ShareTopicFilter}),
             RankProgress = emqx_ds_shared_sub_leader_rank_progress:init(),
             Store0 = emqx_ds_shared_sub_leader_store:init(Group),
             Store1 = emqx_ds_shared_sub_leader_store:set(start_time, StartTime, Store0),
@@ -162,7 +162,7 @@ init_claim_renewal(_Data = #{leader_claim := Claim}) ->
 %%--------------------------------------------------------------------
 %% repalying state
 handle_event(enter, _OldState, ?leader_active, #{topic := Topic} = _Data) ->
-    ?tp(warning, shared_sub_leader_enter_actve, #{topic => Topic}),
+    ?tp(debug, shared_sub_leader_enter_actve, #{topic => Topic}),
     {keep_state_and_data, [
         {{timeout, #renew_streams{}}, 0, #renew_streams{}},
         {{timeout, #renew_leases{}}, ?dq_config(leader_renew_lease_interval_ms), #renew_leases{}},
@@ -407,7 +407,7 @@ revoke_excess_streams_from_agent(Data0, Agent, DesiredCount) ->
             false ->
                 AgentState0;
             true ->
-                ?tp(warning, shared_sub_leader_revoke_streams, #{
+                ?tp(debug, shared_sub_leader_revoke_streams, #{
                     agent => Agent,
                     agent_stream_count => length(Streams0),
                     revoke_count => RevokeCount,
@@ -460,7 +460,7 @@ assign_lacking_streams(Data0, Agent, DesiredCount) ->
         false ->
             Data0;
         true ->
-            ?tp(warning, shared_sub_leader_assign_streams, #{
+            ?tp(debug, shared_sub_leader_assign_streams, #{
                 agent => Agent,
                 agent_stream_count => length(Streams0),
                 assign_count => AssignCount,
@@ -488,7 +488,7 @@ select_streams_for_assign(Data0, _Agent, AssignCount) ->
 %% renew_leases - send lease confirmations to agents
 
 renew_leases(#{agents := AgentStates} = Data) ->
-    ?tp(warning, shared_sub_leader_renew_leases, #{agents => maps:keys(AgentStates)}),
+    ?tp(debug, shared_sub_leader_renew_leases, #{agents => maps:keys(AgentStates)}),
     ok = lists:foreach(
         fun({Agent, AgentState}) ->
             renew_lease(Data, Agent, AgentState)
@@ -562,7 +562,7 @@ connect_agent(
     }),
     case Agents of
         #{Agent := AgentState} ->
-            ?tp(warning, shared_sub_leader_agent_already_connected, #{
+            ?tp(debug, shared_sub_leader_agent_already_connected, #{
                 agent => Agent
             }),
             reconnect_agent(Data, Agent, AgentMetadata, AgentState);
@@ -585,7 +585,7 @@ reconnect_agent(
     AgentMetadata,
     #{streams := OldStreams, revoked_streams := OldRevokedStreams} = _OldAgentState
 ) ->
-    ?tp(warning, shared_sub_leader_agent_reconnect, #{
+    ?tp(debug, shared_sub_leader_agent_reconnect, #{
         agent => Agent,
         agent_metadata => AgentMetadata,
         inherited_streams => OldStreams
@@ -820,7 +820,7 @@ update_agent_stream_states(Data0, Agent, AgentStreamProgresses, VersionOld, Vers
 disconnect_agent(Data0, Agent, AgentStreamProgresses, Version) ->
     case get_agent_state(Data0, Agent) of
         #{version := Version} ->
-            ?tp(warning, shared_sub_leader_disconnect_agent, #{
+            ?tp(debug, shared_sub_leader_disconnect_agent, #{
                 agent => Agent,
                 version => Version
             }),
@@ -847,7 +847,7 @@ agent_transition_to_waiting_updating(
     Streams,
     RevokedStreams
 ) ->
-    ?tp(warning, shared_sub_leader_agent_state_transition, #{
+    ?tp(debug, shared_sub_leader_agent_state_transition, #{
         agent => Agent,
         old_state => OldState,
         new_state => ?waiting_updating
@@ -871,7 +871,7 @@ agent_transition_to_waiting_updating(
 agent_transition_to_waiting_replaying(
     #{group_id := GroupId} = _Data, Agent, #{state := OldState, version := Version} = AgentState0
 ) ->
-    ?tp(warning, shared_sub_leader_agent_state_transition, #{
+    ?tp(debug, shared_sub_leader_agent_state_transition, #{
         agent => Agent,
         old_state => OldState,
         new_state => ?waiting_replaying
@@ -886,7 +886,7 @@ agent_transition_to_waiting_replaying(
 agent_transition_to_initial_waiting_replaying(
     #{group_id := GroupId} = Data, Agent, AgentMetadata, InitialStreams
 ) ->
-    ?tp(warning, shared_sub_leader_agent_state_transition, #{
+    ?tp(debug, shared_sub_leader_agent_state_transition, #{
         agent => Agent,
         old_state => none,
         new_state => ?waiting_replaying
@@ -909,7 +909,7 @@ agent_transition_to_initial_waiting_replaying(
     renew_no_replaying_deadline(AgentState).
 
 agent_transition_to_replaying(Agent, #{state := ?waiting_replaying} = AgentState) ->
-    ?tp(warning, shared_sub_leader_agent_state_transition, #{
+    ?tp(debug, shared_sub_leader_agent_state_transition, #{
         agent => Agent,
         old_state => ?waiting_replaying,
         new_state => ?replaying
@@ -921,7 +921,7 @@ agent_transition_to_replaying(Agent, #{state := ?waiting_replaying} = AgentState
     }.
 
 agent_transition_to_updating(Agent, #{state := ?waiting_updating} = AgentState0) ->
-    ?tp(warning, shared_sub_leader_agent_state_transition, #{
+    ?tp(debug, shared_sub_leader_agent_state_transition, #{
         agent => Agent,
         old_state => ?waiting_updating,
         new_state => ?updating
@@ -1047,7 +1047,7 @@ drop_agent(#{agents := Agents} = Data0, Agent) ->
     #{streams := Streams, revoked_streams := RevokedStreams} = AgentState,
     AllStreams = Streams ++ RevokedStreams,
     Data1 = unassign_streams(Data0, AllStreams),
-    ?tp(warning, shared_sub_leader_drop_agent, #{agent => Agent}),
+    ?tp(debug, shared_sub_leader_drop_agent, #{agent => Agent}),
     Data1#{agents => maps:remove(Agent, Agents)}.
 
 invalidate_agent(#{group_id := GroupId}, Agent) ->