From dbd9ede67b1dfea74a02320a4bacea2643e3341b Mon Sep 17 00:00:00 2001 From: David Ansari Date: Wed, 30 Oct 2024 14:50:05 +0100 Subject: [PATCH] Use log macros for AMQP Using a log macro has the benefit that location data is added as explained in https://www.erlang.org/doc/apps/kernel/logger.html#t:metadata/0 --- deps/rabbit/src/rabbit_amqp_reader.erl | 41 ++++++++++++------------- deps/rabbit/src/rabbit_amqp_session.erl | 23 +++++++------- 2 files changed, 32 insertions(+), 32 deletions(-) diff --git a/deps/rabbit/src/rabbit_amqp_reader.erl b/deps/rabbit/src/rabbit_amqp_reader.erl index 9ae1c3e6ee..070205fa0b 100644 --- a/deps/rabbit/src/rabbit_amqp_reader.erl +++ b/deps/rabbit/src/rabbit_amqp_reader.erl @@ -7,6 +7,7 @@ -module(rabbit_amqp_reader). +-include_lib("kernel/include/logger.hrl"). -include_lib("rabbit_common/include/rabbit.hrl"). -include_lib("amqp10_common/include/amqp10_types.hrl"). -include("rabbit_amqp.hrl"). @@ -329,16 +330,14 @@ error_frame(Condition, Fmt, Args) -> handle_exception(State = #v1{connection_state = closed}, Channel, #'v1_0.error'{description = {utf8, Desc}}) -> - rabbit_log_connection:error( - "Error on AMQP 1.0 connection ~tp (~tp), channel number ~b:~n~tp", - [self(), closed, Channel, Desc]), + ?LOG_ERROR("Error on AMQP 1.0 connection ~tp (~tp), channel number ~b:~n~tp", + [self(), closed, Channel, Desc]), State; handle_exception(State = #v1{connection_state = CS}, Channel, Error = #'v1_0.error'{description = {utf8, Desc}}) when ?IS_RUNNING(State) orelse CS =:= closing -> - rabbit_log_connection:error( - "Error on AMQP 1.0 connection ~tp (~tp), channel number ~b:~n~tp", - [self(), CS, Channel, Desc]), + ?LOG_ERROR("Error on AMQP 1.0 connection ~tp (~tp), channel number ~b:~n~tp", + [self(), CS, Channel, Desc]), close(Error, State); handle_exception(State, _Channel, Error) -> silent_close_delay(), @@ -438,10 +437,10 @@ handle_connection_frame( Timer = maybe_start_credential_expiry_timer(User), rabbit_core_metrics:auth_attempt_succeeded(<<>>, Username, amqp10), notify_auth(user_authentication_success, Username, State0), - rabbit_log_connection:info( - "Connection from AMQP 1.0 container '~ts': user '~ts' authenticated " - "using SASL mechanism ~s and granted access to vhost '~ts'", - [ContainerId, Username, Mechanism, Vhost]), + ?LOG_INFO( + "Connection from AMQP 1.0 container '~ts': user '~ts' authenticated " + "using SASL mechanism ~s and granted access to vhost '~ts'", + [ContainerId, Username, Mechanism, Vhost]), OutgoingMaxFrameSize = case ClientMaxFrame of undefined -> @@ -519,9 +518,9 @@ handle_connection_frame( null -> undefined; {utf8, Val} -> Val end, - rabbit_log:debug( - "AMQP 1.0 connection.open frame: hostname = ~ts, extracted vhost = ~ts, idle-time-out = ~p", - [HostnameVal, Vhost, IdleTimeout]), + ?LOG_DEBUG( + "AMQP 1.0 connection.open frame: hostname = ~ts, extracted vhost = ~ts, idle-time-out = ~p", + [HostnameVal, Vhost, IdleTimeout]), Infos = infos(?CONNECTION_EVENT_KEYS, State), ok = rabbit_core_metrics:connection_created( @@ -780,16 +779,16 @@ notify_auth(EventType, Username, State) -> rabbit_event:notify(EventType, EventProps). track_channel(ChannelNum, SessionPid, #v1{tracked_channels = Channels} = State) -> - rabbit_log:debug("AMQP 1.0 created session process ~p for channel number ~b", - [SessionPid, ChannelNum]), + ?LOG_DEBUG("AMQP 1.0 created session process ~p for channel number ~b", + [SessionPid, ChannelNum]), _Ref = erlang:monitor(process, SessionPid, [{tag, {'DOWN', ChannelNum}}]), State#v1{tracked_channels = maps:put(ChannelNum, SessionPid, Channels)}. untrack_channel(ChannelNum, SessionPid, #v1{tracked_channels = Channels0} = State) -> case maps:take(ChannelNum, Channels0) of {SessionPid, Channels} -> - rabbit_log:debug("AMQP 1.0 closed session process ~p with channel number ~b", - [SessionPid, ChannelNum]), + ?LOG_DEBUG("AMQP 1.0 closed session process ~p with channel number ~b", + [SessionPid, ChannelNum]), State#v1{tracked_channels = Channels}; _ -> State @@ -890,7 +889,7 @@ set_credential0(Cred, credential_timer = OldTimer} = Conn, tracked_channels = Chans, sock = Sock}) -> - rabbit_log:info("updating credential", []), + ?LOG_INFO("updating credential", []), case rabbit_access_control:update_state(User0, Cred) of {ok, User} -> try rabbit_access_control:check_vhost_access(User, Vhost, {socket, Sock}, #{}) of @@ -925,9 +924,9 @@ maybe_start_credential_expiry_timer(User) -> undefined; Ts when is_integer(Ts) -> Time = (Ts - os:system_time(second)) * 1000, - rabbit_log:debug( - "credential expires in ~b ms frow now (absolute timestamp = ~b seconds since epoch)", - [Time, Ts]), + ?LOG_DEBUG( + "credential expires in ~b ms frow now (absolute timestamp = ~b seconds since epoch)", + [Time, Ts]), case Time > 0 of true -> erlang:send_after(Time, self(), credential_expired); diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl index a406de7c42..8e965aa8c8 100644 --- a/deps/rabbit/src/rabbit_amqp_session.erl +++ b/deps/rabbit/src/rabbit_amqp_session.erl @@ -11,6 +11,7 @@ -behaviour(gen_server). +-include_lib("kernel/include/logger.hrl"). -include_lib("rabbit_common/include/rabbit.hrl"). -include_lib("amqp10_common/include/amqp10_types.hrl"). -include("rabbit_amqp.hrl"). @@ -601,8 +602,8 @@ log_error_and_close_session( writer_pid = WriterPid, channel_num = Ch}}) -> End = #'v1_0.end'{error = Error}, - rabbit_log:warning("Closing session for connection ~p: ~tp", - [ReaderPid, Error]), + ?LOG_WARNING("Closing session for connection ~p: ~tp", + [ReaderPid, Error]), ok = rabbit_amqp_writer:send_command_sync(WriterPid, Ch, End), {stop, {shutdown, Error}, State}. @@ -889,8 +890,8 @@ destroy_outgoing_link(_, _, _, Acc) -> Acc. detach(Handle, Link, Error = #'v1_0.error'{}) -> - rabbit_log:warning("Detaching link handle ~b due to error: ~tp", - [Handle, Error]), + ?LOG_WARNING("Detaching link handle ~b due to error: ~tp", + [Handle, Error]), publisher_or_consumer_deleted(Link), #'v1_0.detach'{handle = ?UINT(Handle), closed = true, @@ -981,8 +982,8 @@ handle_frame(#'v1_0.flow'{handle = Handle} = Flow, %% "If set to a handle that is not currently associated with %% an attached link, the recipient MUST respond by ending the %% session with an unattached-handle session error." [2.7.4] - rabbit_log:warning( - "Received Flow frame for unknown link handle: ~tp", [Flow]), + ?LOG_WARNING("Received Flow frame for unknown link handle: ~tp", + [Flow]), protocol_error( ?V_1_0_SESSION_ERROR_UNATTACHED_HANDLE, "Unattached link handle: ~b", [HandleInt]) @@ -2161,9 +2162,9 @@ handle_deliver(ConsumerTag, AckRequired, outgoing_links = OutgoingLinks}; _ -> %% TODO handle missing link -- why does the queue think it's there? - rabbit_log:warning( - "No link handle ~b exists for delivery with consumer tag ~p from queue ~tp", - [Handle, ConsumerTag, QName]), + ?LOG_WARNING( + "No link handle ~b exists for delivery with consumer tag ~p from queue ~tp", + [Handle, ConsumerTag, QName]), State end. @@ -3008,7 +3009,7 @@ credit_reply_timeout(QType, QName) -> Fmt = "Timed out waiting for credit reply from ~s ~s. " "Hint: Enable feature flag rabbitmq_4.0.0", Args = [QType, rabbit_misc:rs(QName)], - rabbit_log:error(Fmt, Args), + ?LOG_ERROR(Fmt, Args), protocol_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR, Fmt, Args). default(undefined, Default) -> Default; @@ -3547,7 +3548,7 @@ recheck_authz(#state{incoming_links = IncomingLinks, permission_cache = Cache0, cfg = #cfg{user = User} } = State) -> - rabbit_log:debug("rechecking link authorizations", []), + ?LOG_DEBUG("rechecking link authorizations", []), Cache1 = maps:fold( fun(_Handle, #incoming_link{exchange = X}, Cache) -> case X of