TiKV 源码解析系列文章（四）Prometheus（下）

pub type HistogramVec = MetricVec<HistogramVecBuilder>;

pub struct MetricVec<T: MetricVecBuilder> {
   pub(crate) v: Arc<MetricVecCore<T>>,
}

impl<T: MetricVecBuilder> MetricVec<T> {
   pub fn with_label_values(&self, vals: &[&str]) -> T::M {
       self.get_metric_with_label_values(vals).unwrap()
   }
}

pub trait MetricVecBuilder: Send + Sync + Clone {
   type M: Metric;
   type P: Describer + Sync + Send + Clone;

   fn build(&self, &Self::P, &[&str]) -> Result<Self::M>;
}

pub(crate) struct MetricVecCore<T: MetricVecBuilder> {
   pub children: RwLock<HashMap<u64, T::M>>,
   // Some fields are omitted.
}

impl<T: MetricVecBuilder> MetricVecCore<T> {
   // Some functions are omitted.

   pub fn get_metric_with_label_values(&self, vals: &[&str]) -> Result<T::M> {
       let h = self.hash_label_values(vals)?;

       if let Some(metric) = self.children.read().get(&h).cloned() {
           return Ok(metric);
       }

       self.get_or_create_metric(h, vals)
   }

   pub(crate) fn hash_label_values(&self, vals: &[&str]) -> Result<u64> {
       if vals.len() != self.desc.variable_labels.len() {
           return Err(Error::InconsistentCardinality(
               self.desc.variable_labels.len(),
               vals.len(),
           ));
       }

       let mut h = FnvHasher::default();
       for val in vals {
           h.write(val.as_bytes());
       }

       Ok(h.finish())
   }

   fn get_or_create_metric(&self, hash: u64, label_values: &[&str]) -> Result<T::M> {
       let mut children = self.children.write();
       // Check exist first.
       if let Some(metric) = children.get(&hash).cloned() {
           return Ok(metric);
       }

       let metric = self.new_metric.build(&self.opts, label_values)?;
       children.insert(hash, metric.clone());
       Ok(metric)
   }
}

pub trait Atomic: Send + Sync {
   /// The numeric type associated with this atomic.
   type T: Number;
   /// Create a new atomic value.
   fn new(val: Self::T) -> Self;
   /// Set the value to the provided value.
   fn set(&self, val: Self::T);
   /// Get the value.
   fn get(&self) -> Self::T;
   /// Increment the value by a given amount.
   fn inc_by(&self, delta: Self::T);
   /// Decrement the value by a given amount.
   fn dec_by(&self, delta: Self::T);
}

pub struct Value<P: Atomic> {
   pub val: P,
   // Some fields are omitted.
}

pub struct GenericCounter<P: Atomic> {
   v: Arc<Value<P>>,
}

pub type Counter = GenericCounter<AtomicF64>;
pub type IntCounter = GenericCounter<AtomicI64>;

pub struct GenericLocalCounter<P: Atomic> {
   counter: GenericCounter<P>,
   val: P::T,
}

pub type LocalCounter = GenericLocalCounter<AtomicF64>;
pub type LocalIntCounter = GenericLocalCounter<AtomicI64>;

impl<P: Atomic> GenericLocalCounter<P> {
   // Some functions are omitted.

   pub fn flush(&mut self) {
       if self.val == P::T::from_i64(0) {
           return;
       }
       self.counter.inc_by(self.val);
       self.val = P::T::from_i64(0);
   }
}

pub struct LocalHistogramCore {
   histogram: Histogram,
   counts: Vec<u64>,
   count: u64,
   sum: f64,
}

impl LocalHistogramCore {
   // Some functions are omitted.

   pub fn observe(&mut self, v: f64) {
       // Try find the bucket.
       let mut iter = self
           .histogram
           .core
           .upper_bounds
           .iter()
           .enumerate()
           .filter(|&(_, f)| v <= *f);
       if let Some((i, _)) = iter.next() {
           self.counts[i] += 1;
       }

       self.count += 1;
       self.sum += v;
   }

   pub fn flush(&mut self) {
       // No cached metric, return.
       if self.count == 0 {
           return;
       }
       {
           let h = &self.histogram;
           for (i, v) in self.counts.iter().enumerate() {
               if *v > 0 {
                   h.core.counts[i].inc_by(*v);
               }
           }
           h.core.count.inc_by(self.count);
           h.core.sum.inc_by(self.sum);
       }
       self.clear();
   }
}

struct Metrics {
   kv_get: Histogram,
   kv_scan: Histogram,
   kv_prewrite: Histogram,
   kv_commit: Histogram,
   // ...
}

impl Metrics {
   fn new() -> Metrics {
       Metrics {
           kv_get: GRPC_MSG_HISTOGRAM_VEC.with_label_values(&["kv_get"]),
           kv_scan: GRPC_MSG_HISTOGRAM_VEC.with_label_values(&["kv_scan"]),
           kv_prewrite: GRPC_MSG_HISTOGRAM_VEC.with_label_values(&["kv_prewrite"]),
           kv_commit: GRPC_MSG_HISTOGRAM_VEC.with_label_values(&["kv_commit"]),
           // ...
       }
   }
}

@@ -102,10 +155,8 @@ fn make_callback<T: Debug + Send + 'static>() -> (Box<FnBox(T) + Send>, oneshot:

impl<T: RaftStoreRouter + 'static> tikvpb_grpc::Tikv for Service<T> {
    fn kv_get(&self, ctx: RpcContext, mut req: GetRequest, sink: UnarySink<GetResponse>) {
-        let label = "kv_get";
-        let timer = GRPC_MSG_HISTOGRAM_VEC
-            .with_label_values(&[label])
-            .start_coarse_timer();
+        const LABEL: &str = "kv_get";
+        let timer = self.metrics.kv_get.start_coarse_timer();

        let (cb, future) = make_callback();
        let res = self.storage.async_get(

make_static_metric! {
   pub struct GrpcMsgHistogram: Histogram {
       "type" => {
           kv_get,
           kv_scan,
           kv_prewrite,
           kv_commit,
           // ...
       },
   }
}

let metrics = GrpcMsgHistogram::from(GRPC_MSG_HISTOGRAM_VEC);

// Usage:
metrics.kv_get.start_coarse_timer();