chore: add more info logging to semaphore (#17831)

Author: drmingdrmer

src/meta/semaphore/src/acquirer/acquirer.rs

@@ -150,7 +150,11 @@ impl Acquirer {
         // Step 4: Wait for the semaphore to be acquired or removed.
 
         while let Some(sem_event) = self.permit_event_rx.recv().await {
-            debug!("semaphore event: {:?}", sem_event);
+            info!(
+                "Acquirer({}): received semaphore event: {:?}",
+                self.ctx, sem_event
+            );
+
             match sem_event {
                 PermitEvent::Acquired((seq, _)) => {
                     if seq == permit_key.seq {

src/meta/semaphore/src/meta_event_subscriber/processor.rs

@@ -15,6 +15,7 @@
 use codeq::Decode;
 use databend_common_meta_types::protobuf::WatchResponse;
 use databend_common_meta_types::SeqV;
+use log::info;
 use log::warn;
 use tokio::sync::mpsc;
 
@@ -74,6 +75,11 @@ impl Processor {
         prev: Option<SeqV<PermitEntry>>,
         current: Option<SeqV<PermitEntry>>,
     ) -> Result<(), ConnectionClosed> {
+        info!(
+            "{} processing kv change: {}: {:?} -> {:?}",
+            self.ctx, sem_key, prev, current
+        );
+
         // Update local queue to update the acquired/released state.
         let state_changes = match (prev, current) {
             (None, Some(entry)) => self.queue.insert(sem_key.seq, entry.data),
@@ -89,7 +95,11 @@ impl Processor {
             }
         };
 
+        info!("{} queue state: {}", self.ctx, self.queue);
+
         for event in state_changes {
+            info!("{} sending event: {}", self.ctx, event);
+
             self.tx.send(event).await.map_err(|e| {
                 ConnectionClosed::new_str(format!("Semaphore-Watcher fail to send {}", e.0))
                     .context(&self.ctx)

src/meta/semaphore/src/meta_event_subscriber/subscriber.rs

@@ -74,6 +74,11 @@ impl MetaEventSubscriber {
                 .context(&self.ctx)
         })?;
 
+        info!(
+            "{} watch stream created: [{}, {})",
+            self.ctx, self.left, self.right
+        );
+
         Ok(strm)
     }
 
@@ -116,6 +121,11 @@ impl MetaEventSubscriber {
                 }
             };
 
+            info!(
+                "{} received event from watch-stream: {:?}",
+                self.ctx, watch_result
+            );
+
             let Some(watch_response) = watch_result? else {
                 // TODO: add retry connecting.
                 error!("watch-stream closed: {}", self.ctx);

src/meta/semaphore/src/queue/semaphore_queue.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::fmt;
 
 use crate::queue::semaphore_event::PermitEvent;
 use crate::PermitEntry;
@@ -32,6 +33,27 @@ pub struct SemaphoreQueue {
     waiting: BTreeMap<PermitSeq, PermitEntry>,
 }
 
+impl fmt::Display for SemaphoreQueue {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SemaphoreQueue{{ {}/{}", self.size, self.capacity,)?;
+
+        write!(f, ", acquired: [")?;
+        for (seq, entry) in &self.acquired {
+            write!(f, "{}:{} ", seq, entry)?;
+        }
+        write!(f, "]")?;
+
+        write!(f, ", waiting: [")?;
+        for (seq, entry) in &self.waiting {
+            write!(f, "{}:{} ", seq, entry)?;
+        }
+        write!(f, "]")?;
+
+        write!(f, "}}")?;
+        Ok(())
+    }
+}
+
 impl SemaphoreQueue {
     pub fn new(capacity: u64) -> Self {
         SemaphoreQueue {
@@ -152,6 +174,21 @@ mod tests {
     use crate::queue::*;
     use crate::PermitEntry;
 
+    #[test]
+    fn test_display() {
+        let queue = SemaphoreQueue {
+            size: 10,
+            capacity: 20,
+            acquired: BTreeMap::from([(1, ent("t1", 3)), (2, ent("t2", 4))]),
+            waiting: BTreeMap::from([(3, ent("t3", 5)), (4, ent("t4", 6))]),
+        };
+
+        assert_eq!(
+            format!("{}", queue),
+            "SemaphoreQueue{ 10/20, acquired: [1:PermitEntry(id:t1, n:3) 2:PermitEntry(id:t2, n:4) ], waiting: [3:PermitEntry(id:t3, n:5) 4:PermitEntry(id:t4, n:6) ]}"
+        );
+    }
+
     #[test]
     fn test_insert() {
         // Test case 1: Insert when there is enough capacity