chore: add logging to semphore leaser (#17825)

* chore(query): add statement_queue_ttl_in_seconds setting for queries queue

* chore: add logging to semphore leaser

* Update acquirer.rs

* Update settings_default.rs

---------

Co-authored-by: zhang2014 <[email protected]>

Author: drmingdrmer

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

@@ -227,6 +227,7 @@ impl Acquirer {
             val_bytes,
             self.lease,
             cancel_rx.map(|_| ()),
+            self.ctx.clone(),
         );
 
         let task_name = format!("{}/(seq={})", self.ctx, sem_key.seq);
@@ -256,7 +257,10 @@ impl Acquirer {
         val_bytes: Vec<u8>,
         ttl: Duration,
         cancel: impl Future<Output = ()> + Send + 'static,
+        ctx: impl ToString,
     ) -> Result<(), ConnectionClosed> {
+        let ctx = ctx.to_string();
+
         let sleep_time = ttl / 3;
         let sleep_time = std::cmp::min(sleep_time, Duration::from_millis(2_000));
 
@@ -287,6 +291,11 @@ impl Acquirer {
             // it means the semaphore has been released. Re-inserting it would cause confusion
             // in the semaphore state and potentially lead to inconsistent behavior.
 
+            info!(
+                "{}: About to extend semaphore permit lease: {} ttl: {:?}",
+                ctx, key_str, ttl
+            );
+
             let upsert = UpsertKV::update(&key_str, &val_bytes)
                 .with(MatchSeq::GE(1))
                 .with_ttl(ttl);

src/query/service/src/sessions/queue_mgr.rs

@@ -66,6 +66,8 @@ pub trait QueueData: Send + Sync + 'static {
 
     fn remove_error_message(key: Option<Self::Key>) -> ErrorCode;
 
+    fn lock_ttl(&self) -> Duration;
+
     fn timeout(&self) -> Duration;
 
     fn need_acquire_to_queue(&self) -> bool;
@@ -158,7 +160,7 @@ impl<Data: QueueData> QueueManager<Data> {
                 data.get_lock_key(),
                 self.permits as u64,
                 data.get_key(), // ID of this acquirer
-                Duration::from_secs(3),
+                data.lock_ttl(),
             );
 
             let future = AcquireQueueFuture::create(
@@ -333,6 +335,7 @@ pub struct QueryEntry {
     pub sql: String,
     pub user_info: UserInfo,
     pub timeout: Duration,
+    pub lock_ttl: Duration,
     pub need_acquire_to_queue: bool,
 }
 
@@ -354,6 +357,7 @@ impl QueryEntry {
                 0 => Duration::from_secs(60 * 60 * 24 * 365 * 35),
                 timeout => Duration::from_secs(timeout),
             },
+            lock_ttl: Duration::from_secs(settings.get_statement_queue_ttl_in_seconds()?),
         })
     }
 
@@ -481,6 +485,10 @@ impl QueueData for QueryEntry {
         }
     }
 
+    fn lock_ttl(&self) -> Duration {
+        self.lock_ttl
+    }
+
     fn timeout(&self) -> Duration {
         self.timeout
     }

src/query/service/tests/it/sessions/queue_mgr.rs

@@ -52,6 +52,10 @@ impl<const PASSED: bool> QueueData for TestData<PASSED> {
         ErrorCode::Internal(format!("{:?}", key))
     }
 
+    fn lock_ttl(&self) -> Duration {
+        Duration::from_secs(3)
+    }
+
     fn timeout(&self) -> Duration {
         Duration::from_secs(1000)
     }

src/query/settings/src/settings_default.rs

@@ -1018,6 +1018,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
+                ("statement_queue_ttl_in_seconds", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(15),
+                    desc: "This parameter specifies the interval, in seconds, between lease renewal operations with the meta service to maintain active communication.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(1..=3600)),
+                }),
                 ("geometry_output_format", DefaultSettingValue {
                     value: UserSettingValue::String("GeoJSON".to_owned()),
                     desc: "Display format for GEOMETRY values.",

src/query/settings/src/settings_getter_setter.rs

@@ -932,4 +932,8 @@ impl Settings {
     pub fn get_enable_block_stream_write(&self) -> Result<bool> {
         Ok(self.try_get_u64("enable_block_stream_write")? == 1)
     }
+
+    pub fn get_statement_queue_ttl_in_seconds(&self) -> Result<u64> {
+        self.try_get_u64("statement_queue_ttl_in_seconds")
+    }
 }