[ISSUE #8092]♻️Structure hot path log fields for RocketMQ services

rocketmq-client/src/consumer/default_lite_pull_consumer.rs

@@ -1132,10 +1132,10 @@ impl LitePullConsumer for DefaultLitePullConsumer {
         match self.try_impl_() {
             Ok(impl_) => {
                 if let Err(e) = impl_.mut_from_ref().unsubscribe(wrapped_topic).await {
-                    tracing::warn!("Failed to unsubscribe from topic {}: {}", topic, e);
+                    tracing::warn!(topic = %topic, error = %e, "unsubscribe failed");
                 }
             }
-            Err(e) => tracing::warn!("Failed to unsubscribe from topic {}: {}", topic, e),
+            Err(e) => tracing::warn!(topic = %topic, error = %e, "unsubscribe failed"),
         }
     }
 
@@ -1191,12 +1191,12 @@ impl LitePullConsumer for DefaultLitePullConsumer {
             Ok(impl_) => match impl_.poll(self.consumer_config.poll_timeout_millis).await {
                 Ok(messages) => messages,
                 Err(e) => {
-                    tracing::error!("Poll failed: {}", e);
+                    tracing::error!(error = %e, "poll failed");
                     Vec::new()
                 }
             },
             Err(e) => {
-                tracing::error!("Poll failed: {}", e);
+                tracing::error!(error = %e, "poll failed");
                 Vec::new()
             }
         }
@@ -1208,12 +1208,12 @@ impl LitePullConsumer for DefaultLitePullConsumer {
             Ok(impl_) => match impl_.poll_with_timeout(timeout).await {
                 Ok(messages) => messages,
                 Err(e) => {
-                    tracing::error!("Poll failed: {}", e);
+                    tracing::error!(timeout_ms = timeout, error = %e, "poll failed");
                     Vec::new()
                 }
             },
             Err(e) => {
-                tracing::error!("Poll failed: {}", e);
+                tracing::error!(timeout_ms = timeout, error = %e, "poll failed");
                 Vec::new()
             }
         }

rocketmq-client/src/producer/producer_impl/default_mq_producer_impl.rs

@@ -2498,7 +2498,11 @@ impl DefaultMQProducerImpl {
                 .await;
             let cost = begin_timestamp.elapsed().as_millis() as u64;
             if cost > 500 {
-                warn!("prepare send request for <{}> cost {} ms", msg.topic(), cost);
+                warn!(
+                    topic = %msg.topic(),
+                    elapsed_ms = cost,
+                    "prepare send request slow"
+                );
             }
         }
         Ok(())

rocketmq-remoting/src/clients/rocketmq_tokio_client.rs

@@ -1267,7 +1267,11 @@ impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqD
                         pool.record_success(addr, latency_ms);
                     }
 
-                    debug!("Request to {} completed in {:?}", addr, latency);
+                    debug!(
+                        remote_addr = %addr,
+                        elapsed_ms = latency.as_millis() as u64,
+                        "request completed"
+                    );
                 }
                 Ok(response)
             }
@@ -1279,7 +1283,12 @@ impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqD
                         pool.record_error(addr);
                     }
 
-                    warn!("Request to {} failed after {:?}: {:?}", addr, latency, err);
+                    warn!(
+                        remote_addr = %addr,
+                        elapsed_ms = latency.as_millis() as u64,
+                        error = ?err,
+                        "request failed"
+                    );
                 }
                 Err(err)
             }
@@ -1303,7 +1312,7 @@ impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqD
         let client = self.get_and_create_client(Some(addr)).await;
         match client {
             None => {
-                error!("invokeOneway: get client for {} failed", addr);
+                error!(remote_addr = %addr, "invoke oneway client unavailable");
             }
             Some(mut client) => {
                 let mut request = request;
@@ -1314,13 +1323,17 @@ impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqD
                         .cmd_handler
                         .do_before_rpc_hooks_with_addr(remote_address, Some(&mut request))
                     {
-                        warn!("invokeOneway: before RPC hook failed for {}: {:?}", addr, error);
+                        warn!(
+                            remote_addr = %addr,
+                            error = ?error,
+                            "invoke oneway before RPC hook failed"
+                        );
                         return;
                     }
                 }
                 let addr_clone = addr.clone();
                 let Some(task_group) = self.get_or_create_worker_task_group() else {
-                    warn!("invokeOneway: client is shut down, skipping send to {}", addr);
+                    warn!(remote_addr = %addr, "invoke oneway skipped because client is shut down");
                     return;
                 };
                 if let Err(error) = task_group.spawn_service("remoting.client.oneway-send", async move {
@@ -1333,17 +1346,26 @@ impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqD
                     {
                         Ok(Ok(())) => {}
                         Ok(Err(e)) => {
-                            warn!("invokeOneway: send request to {} failed: {:?}", addr_clone, e);
+                            warn!(
+                                remote_addr = %addr_clone,
+                                error = ?e,
+                                "invoke oneway send failed"
+                            );
                         }
                         Err(_) => {
                             warn!(
-                                "invokeOneway: send request to {} timeout ({}ms)",
-                                addr_clone, timeout_millis
+                                remote_addr = %addr_clone,
+                                timeout_ms = timeout_millis,
+                                "invoke oneway send timed out"
                             );
                         }
                     }
                 }) {
-                    warn!("invokeOneway: failed to spawn send task for {}: {:?}", addr, error);
+                    warn!(
+                        remote_addr = %addr,
+                        error = ?error,
+                        "invoke oneway send task spawn failed"
+                    );
                 }
             }
         }

rocketmq-store/src/log_file/mapped_file/default_mapped_file_impl.rs

@@ -933,10 +933,10 @@ impl MappedFile for DefaultMappedFile {
         MappedFile::shutdown(self, interval_forcibly);
         if self.is_cleanup_over() {
             if let Err(e) = fs::remove_file(self.file_name.as_str()) {
-                error!("delete file failed: {:?}", e);
+                error!(file_name = %self.file_name, error = ?e, "delete file failed");
                 false
             } else {
-                info!("delete file success: {}", self.file_name);
+                info!(file_name = %self.file_name, "delete file success");
                 true
             }
         } else {
@@ -1016,14 +1016,14 @@ impl MappedFile for DefaultMappedFile {
     #[inline]
     fn mlock(&self) {
         if let Err(error) = lock_memory(self.get_mapped_file().as_ptr(), self.file_size as usize) {
-            warn!("Failed to mlock mapped file {}: {}", self.file_name, error);
+            warn!(file_name = %self.file_name, error = %error, "failed to mlock mapped file");
         }
     }
 
     #[inline]
     fn munlock(&self) {
         if let Err(error) = unlock_memory(self.get_mapped_file().as_ptr(), self.file_size as usize) {
-            warn!("Failed to munlock mapped file {}: {}", self.file_name, error);
+            warn!(file_name = %self.file_name, error = %error, "failed to munlock mapped file");
         }
     }