fix(meta-service): watcher metrics should be updated when sender is created and dropped (#18182)

* refactor(meta-service): remove direct `KVApi` impl on `MetaNode`

This commit extracts the implementation of `KVApi` to a standalone
struct.

* fix(meta-service): watcher metrics should be updated when sender is created and dropped

But not when it is added to or removed from the `Dispatcher`.
Because a sender can be removed more than once, since internally the
dispatcher is a span-map, it stores multiple instance of a sender.

- See: databendlabs/watcher#7

* M  src/meta/service/src/meta_service/meta_node_kv_api_impl.rs

Author: drmingdrmer

Cargo.lock

@@ -532,7 +532,7 @@ url = "2.5.4"
 uuid = { version = "1.10.0", features = ["std", "serde", "v4", "v7"] }
 volo-thrift = "0.10"
 walkdir = "2.3.2"
-watcher = { version = "0.4.0" }
+watcher = { version = "0.4.1" }
 wiremock = "0.6"
 wkt = "0.11.1"
 xorf = { version = "0.11.0", default-features = false, features = ["binary-fuse"] }
@@ -658,5 +658,5 @@ sub-cache = { git = "https://github.com/databendlabs/sub-cache", tag = "v0.2.1"
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }
 tantivy-common = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370", package = "tantivy-common" }
 tantivy-jieba = { git = "https://github.com/datafuse-extras/tantivy-jieba", rev = "0e300e9" }
-watcher = { git = "https://github.com/databendlabs/watcher", tag = "v0.4.0" }
+watcher = { git = "https://github.com/databendlabs/watcher", tag = "v0.4.1" }
 xorfilter-rs = { git = "https://github.com/datafuse-extras/xorfilter", tag = "databend-alpha.4" }

Cargo.toml

@@ -145,6 +145,7 @@ impl MetaServiceImpl {
         let reply = match &req {
             MetaGrpcReq::UpsertKV(a) => {
                 let res = m
+                    .kv_api()
                     .upsert_kv(a.clone())
                     .log_elapsed_info(format!("UpsertKV: {:?}", a))
                     .await;

src/meta/service/src/api/grpc/grpc_service.rs

@@ -93,6 +93,7 @@ use crate::meta_service::errors::grpc_error_to_network_err;
 use crate::meta_service::forwarder::MetaForwarder;
 use crate::meta_service::meta_leader::MetaLeader;
 use crate::meta_service::meta_node_kv_api_impl::MetaKVApi;
+use crate::meta_service::meta_node_kv_api_impl::MetaKVApiOwned;
 use crate::meta_service::meta_node_status::MetaNodeStatus;
 use crate::meta_service::watcher::DispatcherHandle;
 use crate::meta_service::watcher::WatchTypes;
@@ -1199,6 +1200,10 @@ impl MetaNode {
     pub fn kv_api(&self) -> MetaKVApi {
         MetaKVApi::new(self)
     }
+
+    pub fn kv_api_owned(self: &Arc<Self>) -> MetaKVApiOwned {
+        MetaKVApiOwned::new(self.clone())
+    }
 }
 
 pub(crate) fn event_filter_from_filter_type(filter_type: FilterType) -> EventFilter {

src/meta/service/src/meta_service/meta_node.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use databend_common_meta_client::MetaGrpcReadReq;
 use databend_common_meta_kvapi::kvapi;
@@ -117,31 +119,34 @@ impl<'a> kvapi::KVApi for MetaKVApi<'a> {
     }
 }
 
-/// Impl kvapi::KVApi for MetaNode.
-///
-/// Write through raft-log.
-/// Read through local state machine, which may not be consistent.
-/// E.g. Read is not guaranteed to see a write.
+/// A wrapper of MetaNode that implements kvapi::KVApi.
+pub struct MetaKVApiOwned {
+    inner: Arc<MetaNode>,
+}
+
+impl MetaKVApiOwned {
+    pub fn new(inner: Arc<MetaNode>) -> Self {
+        Self { inner }
+    }
+}
+
 #[async_trait]
-impl kvapi::KVApi for MetaNode {
+impl kvapi::KVApi for MetaKVApiOwned {
     type Error = MetaAPIError;
 
     async fn upsert_kv(&self, act: UpsertKV) -> Result<UpsertKVReply, Self::Error> {
-        self.kv_api().upsert_kv(act).await
+        self.inner.kv_api().upsert_kv(act).await
     }
 
-    #[fastrace::trace]
     async fn get_kv_stream(&self, keys: &[String]) -> Result<KVStream<Self::Error>, Self::Error> {
-        self.kv_api().get_kv_stream(keys).await
+        self.inner.kv_api().get_kv_stream(keys).await
     }
 
-    #[fastrace::trace]
     async fn list_kv(&self, prefix: &str) -> Result<KVStream<Self::Error>, Self::Error> {
-        self.kv_api().list_kv(prefix).await
+        self.inner.kv_api().list_kv(prefix).await
     }
 
-    #[fastrace::trace]
     async fn transaction(&self, txn: TxnRequest) -> Result<TxnReply, Self::Error> {
-        self.kv_api().transaction(txn).await
+        self.inner.kv_api().transaction(txn).await
     }
 }

src/meta/service/src/meta_service/meta_node_kv_api_impl.rs

@@ -16,6 +16,9 @@ mod errors;
 mod forwarder;
 mod meta_node_kv_api_impl;
 
+pub use meta_node_kv_api_impl::MetaKVApi;
+pub use meta_node_kv_api_impl::MetaKVApiOwned;
+
 pub(crate) mod snapshot_receiver_v1;
 
 pub mod meta_leader;

src/meta/service/src/meta_service/mod.rs

@@ -17,7 +17,7 @@ use std::sync::Mutex;
 
 use async_trait::async_trait;
 use databend_common_meta_kvapi::kvapi;
-use databend_meta::meta_service::MetaNode;
+use databend_meta::meta_service::MetaKVApiOwned;
 use maplit::btreeset;
 use test_harness::test;
 
@@ -32,8 +32,8 @@ struct MetaNodeUnitTestBuilder {
 }
 
 #[async_trait]
-impl kvapi::ApiBuilder<Arc<MetaNode>> for MetaNodeUnitTestBuilder {
-    async fn build(&self) -> Arc<MetaNode> {
+impl kvapi::ApiBuilder<MetaKVApiOwned> for MetaNodeUnitTestBuilder {
+    async fn build(&self) -> MetaKVApiOwned {
         let (_id, tc) = start_meta_node_leader().await.unwrap();
 
         let meta_node = tc.meta_node();
@@ -43,20 +43,20 @@ impl kvapi::ApiBuilder<Arc<MetaNode>> for MetaNodeUnitTestBuilder {
             tcs.push(tc);
         }
 
-        meta_node
+        meta_node.kv_api_owned()
     }
 
-    async fn build_cluster(&self) -> Vec<Arc<MetaNode>> {
+    async fn build_cluster(&self) -> Vec<MetaKVApiOwned> {
         let (_log_index, tcs) = start_meta_node_cluster(btreeset! {0,1,2}, btreeset! {3,4})
             .await
             .unwrap();
 
         let cluster = vec![
-            tcs[0].meta_node(),
-            tcs[1].meta_node(),
-            tcs[2].meta_node(),
-            tcs[3].meta_node(),
-            tcs[4].meta_node(),
+            tcs[0].meta_node().kv_api_owned(),
+            tcs[1].meta_node().kv_api_owned(),
+            tcs[2].meta_node().kv_api_owned(),
+            tcs[3].meta_node().kv_api_owned(),
+            tcs[4].meta_node().kv_api_owned(),
         ];
 
         {

src/meta/service/tests/it/meta_node/meta_node_kv_api.rs

@@ -68,7 +68,7 @@ async fn test_meta_node_replicate_kv_with_expire() -> anyhow::Result<()> {
 
     info!("--- get kv with expire now+3");
     let seq = {
-        let resp = leader.get_kv(key).await?;
+        let resp = leader.kv_api().get_kv(key).await?;
         let seq_v = resp.unwrap();
         assert_eq!(Some(KVMeta::new_expire(now_sec + 3)), seq_v.meta);
         seq_v.seq
@@ -85,7 +85,7 @@ async fn test_meta_node_replicate_kv_with_expire() -> anyhow::Result<()> {
 
     info!("--- get updated kv with new expire now+1000, assert the updated value");
     {
-        let resp = leader.get_kv(key).await?;
+        let resp = leader.kv_api().get_kv(key).await?;
         let seq_v = resp.unwrap();
         let want = (now_sec + 1000) * 1000;
         let expire_ms = seq_v.meta.unwrap().get_expire_at_ms().unwrap();

src/meta/service/tests/it/meta_node/meta_node_kv_api_expire.rs

@@ -849,7 +849,7 @@ async fn assert_get_kv(
     value: &str,
 ) -> anyhow::Result<()> {
     for (i, mn) in meta_nodes.iter().enumerate() {
-        let got = mn.get_kv(key).await?;
+        let got = mn.kv_api().get_kv(key).await?;
         assert_eq!(
             value.to_string().into_bytes(),
             got.unwrap().data,