Merge branch 'main' into remove-meta-index

Author: mergify[bot]

Cargo.lock

@@ -23,6 +23,9 @@ prometheus-parse = "0.2.3"
 serde = { version = "1.0.137", features = ["derive"] }
 tracing = "0.1.35"
 
+[dev-dependencies]
+anyhow = "1.0.58"
+
 [dev-dependencies.tokio]
 default-features = false
 features = ["io-util", "net", "sync", "rt-multi-thread", "macros"]

src/common/metrics/Cargo.toml

@@ -0,0 +1,132 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! This mod provides mechanism to track the count of active instances of some type `T`.
+//! The count is maintained by a `Count` implementation and will be increased or decreased when a wrapper of `T` `WithCounter` is created or dropped.
+//!
+//! Example:
+//!
+//! ```ignore
+//! struct Connection{}
+//! impl Connection {
+//!     fn ping() {}
+//! }
+//!
+//! struct MyCounter{ identifier: String, }
+//! impl Count for MyCounter {/*...*/}
+//!
+//! {
+//!     let conn = WithCounter::new(Connection{}, MyCounter{}); // increase count with `MyCounter`
+//!     conn.ping();
+//! } // decrease count with `MyCounter`
+//! ```
+
+use std::ops::Deref;
+use std::ops::DerefMut;
+
+/// Defines how to report counter metrics.
+pub trait Count {
+    fn incr_count(&mut self, n: i64);
+}
+
+/// Binds a counter to a `T`.
+///
+/// It counts the number of instances of `T` with the provided counter `Count`.
+pub struct WithCount<C, T>
+where C: Count
+{
+    counter: C,
+    inner: T,
+}
+
+impl<C, T> WithCount<C, T>
+where C: Count
+{
+    pub fn new(t: T, counter: C) -> Self {
+        let mut s = Self { counter, inner: t };
+        s.counter.incr_count(1);
+        s
+    }
+
+    pub fn counter(&self) -> &C {
+        &self.counter
+    }
+}
+
+/// When being dropped, decreases the count.
+impl<C, T> Drop for WithCount<C, T>
+where C: Count
+{
+    fn drop(&mut self) {
+        self.counter.incr_count(-1);
+    }
+}
+
+/// Let an app use `WithCount` the same as using `T`.
+impl<C, T> Deref for WithCount<C, T>
+where C: Count
+{
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+/// Let an app use `WithCount` the same as using `T`.
+impl<C, T> DerefMut for WithCount<C, T>
+where C: Count
+{
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::AtomicI64;
+    use std::sync::atomic::Ordering;
+    use std::sync::Arc;
+
+    use crate::counter::Count;
+    use crate::counter::WithCount;
+
+    struct Foo {}
+    struct Counter {
+        n: Arc<AtomicI64>,
+    }
+    impl Count for Counter {
+        fn incr_count(&mut self, n: i64) {
+            self.n.fetch_add(n, Ordering::Relaxed);
+        }
+    }
+
+    #[test]
+    fn test_with_count() -> anyhow::Result<()> {
+        let count = Arc::new(AtomicI64::new(0));
+        assert_eq!(0, count.load(Ordering::Relaxed));
+
+        {
+            let _a = WithCount::new(Foo {}, Counter { n: count.clone() });
+            assert_eq!(1, count.load(Ordering::Relaxed));
+            {
+                let _b = WithCount::new(Foo {}, Counter { n: count.clone() });
+                assert_eq!(2, count.load(Ordering::Relaxed));
+            }
+            assert_eq!(1, count.load(Ordering::Relaxed));
+        }
+        assert_eq!(0, count.load(Ordering::Relaxed));
+        Ok(())
+    }
+}

src/common/metrics/src/counter.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod counter;
 mod dump;
 mod recorder;
 

src/common/metrics/src/lib.rs

@@ -34,6 +34,7 @@ common-meta-client = { path = "../client" }
 common-meta-raft-store = { path = "../raft-store" }
 common-meta-sled-store = { path = "../sled-store" }
 common-meta-types = { path = "../types" }
+common-metrics = { path = "../../common/metrics" }
 common-tracing = { path = "../../common/tracing" }
 
 # Github dependencies

src/meta/service/Cargo.toml

@@ -21,6 +21,7 @@ pub mod export;
 pub mod meta_service;
 pub mod metrics;
 pub mod network;
+pub mod raft_client;
 pub mod store;
 pub mod version;
 pub mod watcher;

src/meta/service/src/lib.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::ops::Deref;
-use std::ops::DerefMut;
 use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
@@ -22,9 +20,7 @@ use common_base::containers::ItemManager;
 use common_base::containers::Pool;
 use common_meta_sled_store::openraft;
 use common_meta_sled_store::openraft::MessageSummary;
-use common_meta_types::protobuf::raft_service_client::RaftServiceClient;
 use common_meta_types::protobuf::RaftRequest;
-use common_meta_types::Endpoint;
 use common_meta_types::LogEntry;
 use common_meta_types::NodeId;
 use openraft::async_trait::async_trait;
@@ -40,14 +36,15 @@ use tonic::transport::channel::Channel;
 use tracing::debug;
 use tracing::info;
 
-use crate::metrics::incr_meta_metrics_active_peers;
 use crate::metrics::incr_meta_metrics_fail_connections_to_peer;
 use crate::metrics::incr_meta_metrics_sent_bytes_to_peer;
 use crate::metrics::incr_meta_metrics_sent_failure_to_peer;
 use crate::metrics::incr_meta_metrics_snapshot_send_failures_to_peer;
 use crate::metrics::incr_meta_metrics_snapshot_send_inflights_to_peer;
 use crate::metrics::incr_meta_metrics_snapshot_send_success_to_peer;
 use crate::metrics::sample_meta_metrics_snapshot_sent;
+use crate::raft_client::RaftClient;
+use crate::raft_client::RaftClientApi;
 use crate::store::RaftStore;
 
 struct ChannelManager {}
@@ -70,62 +67,6 @@ impl ItemManager for ChannelManager {
     }
 }
 
-/// Client for raft protocol communication
-pub struct RaftClient {
-    target: NodeId,
-    endpoint: Endpoint,
-    endpoint_str: String,
-    inner: RaftServiceClient<Channel>,
-}
-
-impl Deref for RaftClient {
-    type Target = RaftServiceClient<Channel>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.inner
-    }
-}
-
-impl DerefMut for RaftClient {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.inner
-    }
-}
-
-impl Drop for RaftClient {
-    fn drop(&mut self) {
-        incr_meta_metrics_active_peers(&self.target, &self.endpoint_str, -1);
-    }
-}
-
-impl RaftClient {
-    pub fn new(target: NodeId, endpoint: Endpoint, channel: Channel) -> Self {
-        let endpoint_str = endpoint.to_string();
-
-        debug!(
-            "RaftClient::new: target: {} endpoint: {}",
-            target, endpoint_str
-        );
-
-        incr_meta_metrics_active_peers(&target, &endpoint_str, 1);
-
-        Self {
-            target,
-            endpoint,
-            endpoint_str,
-            inner: RaftServiceClient::new(channel),
-        }
-    }
-
-    pub fn endpoint(&self) -> &Endpoint {
-        &self.endpoint
-    }
-
-    pub fn endpoint_str(&self) -> &str {
-        &self.endpoint_str
-    }
-}
-
 pub struct Network {
     sto: Arc<RaftStore>,
 
@@ -150,7 +91,7 @@ impl Network {
 
         match self.conn_pool.get(&addr).await {
             Ok(channel) => {
-                let client = RaftClient::new(*target, endpoint, channel);
+                let client = RaftClientApi::new(*target, endpoint, channel);
                 debug!("connected: target={}: {}", target, addr);
 
                 Ok(client)

src/meta/service/src/network.rs

@@ -0,0 +1,65 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_meta_types::protobuf::raft_service_client::RaftServiceClient;
+use common_meta_types::Endpoint;
+use common_meta_types::NodeId;
+use common_metrics::counter;
+use tonic::transport::channel::Channel;
+use tracing::debug;
+
+use crate::metrics::incr_meta_metrics_active_peers;
+
+/// A metrics reporter of active raft peers.
+pub struct PeerCounter {
+    target: NodeId,
+    endpoint: Endpoint,
+    endpoint_str: String,
+}
+
+impl counter::Count for PeerCounter {
+    fn incr_count(&mut self, n: i64) {
+        incr_meta_metrics_active_peers(&self.target, &self.endpoint_str, n)
+    }
+}
+
+/// RaftClient is a grpc client bound with a metrics reporter..
+pub type RaftClient = counter::WithCount<PeerCounter, RaftServiceClient<Channel>>;
+
+/// Defines the API of the client to a raft node.
+pub trait RaftClientApi {
+    fn new(target: NodeId, endpoint: Endpoint, channel: Channel) -> Self;
+    fn endpoint(&self) -> &Endpoint;
+}
+
+impl RaftClientApi for RaftClient {
+    fn new(target: NodeId, endpoint: Endpoint, channel: Channel) -> Self {
+        let endpoint_str = endpoint.to_string();
+
+        debug!(
+            "RaftClient::new: target: {} endpoint: {}",
+            target, endpoint_str
+        );
+
+        counter::WithCount::new(RaftServiceClient::new(channel), PeerCounter {
+            target,
+            endpoint,
+            endpoint_str,
+        })
+    }
+
+    fn endpoint(&self) -> &Endpoint {
+        &self.counter().endpoint
+    }
+}

src/meta/service/src/raft_client.rs

@@ -188,5 +188,6 @@ service MetaService {
   rpc MemberList(MemberListRequest) returns (MemberListReply);
 
   // Respond with the information about the client.
+  // Since: 2022-09-09 0.8.30
   rpc GetClientInfo(Empty) returns (ClientInfo);
 }