可用区内资源数量监控

discovery/src/distance.rs

@@ -208,6 +208,9 @@ impl DistanceCalculator {
             }
         }
     }
+    pub fn region(&self) -> &Option<String> {
+        &self.local_region
+    }
 }
 
 pub trait Addr {
@@ -494,3 +497,15 @@ impl BClass for &String {
         b
     }
 }
+
+// 本机的region，优先级：
+// 1. 本机IP对应的region
+// 2. 未知region，返回cnx
+pub fn host_region() -> String {
+    let cal = unsafe { DISTANCE_CALCULATOR.get_unchecked().get() };
+    if let Some(region) = cal.region() {
+        return region.clone();
+    }
+
+    return "cnx".to_string();
+}

endpoint/Cargo.toml

@@ -15,6 +15,7 @@ sharding = { path = "../sharding" }
 log = { path = "../log" }
 rt = { path = "../rt" }
 context = { path = "../context" }
+metrics = { path = "../metrics" }
 procs = { path = "../procs" }
 
 byteorder = "1.4.3"

endpoint/src/redisservice/config.rs

@@ -104,4 +104,7 @@ impl RedisNamespace {
 
         true
     }
+    pub(super) fn resource_type(&self) -> &str {
+        self.basic.resource_type.as_str()
+    }
 }

endpoint/src/redisservice/topo.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use crate::{Builder, Endpoint, Single, Topology};
-use discovery::TopologyWrite;
+use discovery::{distance, TopologyWrite};
 use protocol::{Protocol, RedisFlager, Request, Resource};
 use sharding::distribution::Distribute;
 use sharding::hash::{Hash, HashKey, Hasher};
@@ -262,6 +262,7 @@ where
         for (master_addr, slaves) in addrs {
             assert_ne!(master_addr.len(), 0);
             assert_ne!(slaves.len(), 0);
+            let port = master_addr.port().to_string();
             let master = self.take_or_build(&mut old, &master_addr, self.cfg.timeout_master());
             master.enable_single();
 
@@ -281,6 +282,20 @@ where
                 replicas,
                 self.cfg.basic.region_enabled,
             );
+
+            // 生成端口副本数量监控数据
+            let n: u16 = if self.cfg.basic.region_enabled {
+                shard.len_region() + 10000 // snapshot时>0才有输出，len_region这里加10000作为基准值，保证监控有数据；
+            } else {
+                0 // region功能关闭时，len_region为0，也就是不输出监控数据；可用区从打开到关闭场景，0也要生成监控数据，覆盖已有的同path旧数据
+            };
+            metrics::resource_num_metric(
+                self.cfg.resource_type(),
+                self.cfg.service.as_str(),
+                (host_region() + ":" + port.as_str()).as_str(),
+                n,
+            );
+
             self.shards.push(shard);
         }
         assert_eq!(self.shards.len(), self.cfg.shards_url.len());
@@ -292,6 +307,19 @@ where
         true
     }
 }
+
+// 本机的region，优先级：
+// 1. 启动参数/环境变量的region
+// 2. 通过本机IP计算出来的region
+// TODO 目前仅本文件内被使用，后续可以考虑放在一个公共的地方
+fn host_region() -> String {
+    if let Some(region) = context::get().region() {
+        return region.to_string();
+    }
+
+    distance::host_region()
+}
+
 #[derive(Clone)]
 struct Shard<E> {
     master: (String, E),
@@ -327,6 +355,9 @@ impl<E> Shard<E> {
     fn next(&self, idx: usize, runs: usize) -> (usize, &(String, E)) {
         unsafe { self.slaves.unsafe_next(idx, runs) }
     }
+    pub fn len_region(&self) -> u16 {
+        self.slaves.len_region()
+    }
 }
 impl<E: discovery::Inited> Shard<E> {
     // 1. 主已经初始化

metrics/src/lib.rs

@@ -53,6 +53,17 @@ impl Metric {
             &mut *(self as *const _ as *mut _)
         }
     }
+    pub fn inited(&mut self) -> bool {
+        self.item.inited()
+    }
+    // num类型，若未初始化则尝试初始化；若已经初始化，则值清0
+    pub fn zero_num(&mut self) {
+        if !self.inited() {
+            self.try_inited();
+        } else {
+            self.item.data().zero_num();
+        }
+    }
 }
 impl<T: MetricData + Debug> AddAssign<T> for Metric {
     #[inline]

metrics/src/macros.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 pub trait MetricData: Sized {
     fn incr_to(self, data: &ItemData);
     fn incr_to_cache(self, _id: &Arc<Id>) {}
+    fn zero(self, _data: &ItemData) {}
 }
 
 #[derive(Default, Debug)]
@@ -30,6 +31,13 @@ impl ItemData {
         assert!(self.id.t.is_num());
         self.inner.num.incr(num);
     }
+    #[inline]
+    pub fn zero_num(&self) {
+        assert!(self.id.t.is_num());
+        unsafe {
+            self.inner.num.zero();
+        }
+    }
 }
 use crate::ToNumber;
 impl<T: ToNumber> MetricData for T {
@@ -41,6 +49,10 @@ impl<T: ToNumber> MetricData for T {
     fn incr_to_cache(self, id: &Arc<Id>) {
         crate::register_cache(id, self.int());
     }
+    #[inline]
+    fn zero(self, data: &ItemData) {
+        data.zero_num();
+    }
 }
 use ds::time::Duration;
 impl MetricData for Duration {

metrics/src/types/host.rs

@@ -121,3 +121,14 @@ pub fn decr_task() {
 pub fn set_sockfile_failed(failed_count: usize) {
     SOCKFILE_FAILED.store(failed_count as i64, Relaxed);
 }
+
+// fn unchange_number_metric(region_enable:bool,len_region: u16, port: &str, rsname: &str, region: &str) {
+pub fn resource_num_metric(source: &str, namespace: &str, bip: &str, n: u16) {
+    let path = crate::Path::new(vec![source, namespace, bip]);
+    let mut metric = path.num("region_resource");
+    if metric.inited() {
+        metric.zero_num()
+    };
+
+    metric += n;
+}

metrics/src/types/mod.rs

@@ -6,7 +6,7 @@ mod rtt;
 mod status;
 
 pub(crate) use host::*;
-pub use host::{decr_task, incr_task, set_sockfile_failed};
+pub use host::{decr_task, incr_task, resource_num_metric, set_sockfile_failed};
 pub(crate) use number::*;
 pub(crate) use qps::*;
 pub(crate) use ratio::*;

metrics/src/types/number.rs

@@ -55,6 +55,10 @@ impl Number {
     pub(crate) fn incr(&self, v: i64) {
         self.inner.incr(v);
     }
+    #[inline]
+    pub(crate) fn zero(&self) {
+        self.inner.zero();
+    }
 }
 
 pub trait ToNumber {

sharding/src/select/by_distance.rs

@@ -31,7 +31,8 @@ impl BackendQuota {
 // 2. 其他资源，replicas长度与len_local相同
 #[derive(Clone)]
 pub struct Distance<T> {
-    len_local: u16,
+    len_local: u16,  // 实际使用的local实例数量
+    len_region: u16, // 通过排序计算出的可用区内的实例数量，len_region <= len_local
     backend_quota: bool,
     idx: Arc<AtomicUsize>,
     replicas: Vec<(T, BackendQuota)>,
@@ -40,6 +41,7 @@ impl<T: Addr> Distance<T> {
     pub fn new() -> Self {
         Self {
             len_local: 0,
+            len_region: 0,
             backend_quota: false,
             idx: Default::default(),
             replicas: Vec::new(),
@@ -72,11 +74,10 @@ impl<T: Addr> Distance<T> {
             // 按distance选local
             // 1. 距离小于等于4为local
             // 2. local为0，则全部为local
-            let l = replicas.sort_by_region(
-                Vec::new(),
-                context::get().region(),
-                |d, _| d <= discovery::distance::DISTANCE_VAL_REGION,
-            );
+            let l = replicas.sort_by_region(Vec::new(), context::get().region(), |d, _| {
+                d <= discovery::distance::DISTANCE_VAL_REGION
+            });
+            me.len_region = l as u16; // 可用区内的实例数量
             if l == 0 {
                 log::warn!(
                     "too few instance in region:{} total:{}, {:?}",
@@ -103,6 +104,9 @@ impl<T: Addr> Distance<T> {
 
         me
     }
+    pub fn len_region(&self) -> u16 {
+        self.len_region
+    }
     #[inline]
     pub fn from(replicas: Vec<T>) -> Self {
         Self::with_performance_tuning(replicas, true, false)