| // Copyright 2014 The Prometheus Authors |
| // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0. |
| |
| use std::cell::RefCell; |
| use std::collections::HashMap; |
| use std::convert::From; |
| use std::sync::{ |
| atomic::{AtomicU64 as StdAtomicU64, Ordering}, |
| Arc, Mutex, |
| }; |
| use std::time::{Duration, Instant as StdInstant}; |
| |
| use crate::atomic64::{Atomic, AtomicF64, AtomicU64}; |
| use crate::desc::{Desc, Describer}; |
| use crate::errors::{Error, Result}; |
| use crate::metrics::{Collector, LocalMetric, Metric, Opts}; |
| use crate::proto; |
| use crate::value::make_label_pairs; |
| use crate::vec::{MetricVec, MetricVecBuilder}; |
| |
| /// The default [`Histogram`] buckets. The default buckets are |
| /// tailored to broadly measure the response time (in seconds) of a |
| /// network service. Most likely, however, you will be required to define |
| /// buckets customized to your use case. |
| pub const DEFAULT_BUCKETS: &[f64; 11] = &[ |
| 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, |
| ]; |
| |
| /// Used for the label that defines the upper bound of a |
| /// bucket of a histogram ("le" -> "less or equal"). |
| pub const BUCKET_LABEL: &str = "le"; |
| |
| #[inline] |
| fn check_bucket_label(label: &str) -> Result<()> { |
| if label == BUCKET_LABEL { |
| return Err(Error::Msg( |
| "`le` is not allowed as label name in histograms".to_owned(), |
| )); |
| } |
| |
| Ok(()) |
| } |
| |
| fn check_and_adjust_buckets(mut buckets: Vec<f64>) -> Result<Vec<f64>> { |
| if buckets.is_empty() { |
| buckets = Vec::from(DEFAULT_BUCKETS as &'static [f64]); |
| } |
| |
| for (i, upper_bound) in buckets.iter().enumerate() { |
| if i < (buckets.len() - 1) && *upper_bound >= buckets[i + 1] { |
| return Err(Error::Msg(format!( |
| "histogram buckets must be in increasing \ |
| order: {} >= {}", |
| upper_bound, |
| buckets[i + 1] |
| ))); |
| } |
| } |
| |
| let tail = *buckets.last().unwrap(); |
| if tail.is_sign_positive() && tail.is_infinite() { |
| // The +Inf bucket is implicit. Remove it here. |
| buckets.pop(); |
| } |
| |
| Ok(buckets) |
| } |
| |
| /// A struct that bundles the options for creating a [`Histogram`] metric. It is |
| /// mandatory to set Name and Help to a non-empty string. All other fields are |
| /// optional and can safely be left at their zero value. |
| #[derive(Clone, Debug)] |
| pub struct HistogramOpts { |
| /// A container holding various options. |
| pub common_opts: Opts, |
| |
| /// Defines the buckets into which observations are counted. Each |
| /// element in the slice is the upper inclusive bound of a bucket. The |
| /// values must be sorted in strictly increasing order. There is no need |
| /// to add a highest bucket with +Inf bound, it will be added |
| /// implicitly. The default value is DefBuckets. |
| pub buckets: Vec<f64>, |
| } |
| |
| impl HistogramOpts { |
| /// Create a [`HistogramOpts`] with the `name` and `help` arguments. |
| pub fn new<S1: Into<String>, S2: Into<String>>(name: S1, help: S2) -> HistogramOpts { |
| HistogramOpts { |
| common_opts: Opts::new(name, help), |
| buckets: Vec::from(DEFAULT_BUCKETS as &'static [f64]), |
| } |
| } |
| |
| /// `namespace` sets the namespace. |
| pub fn namespace<S: Into<String>>(mut self, namespace: S) -> Self { |
| self.common_opts.namespace = namespace.into(); |
| self |
| } |
| |
| /// `subsystem` sets the sub system. |
| pub fn subsystem<S: Into<String>>(mut self, subsystem: S) -> Self { |
| self.common_opts.subsystem = subsystem.into(); |
| self |
| } |
| |
| /// `const_labels` sets the const labels. |
| pub fn const_labels(mut self, const_labels: HashMap<String, String>) -> Self { |
| self.common_opts = self.common_opts.const_labels(const_labels); |
| self |
| } |
| |
| /// `const_label` adds a const label. |
| pub fn const_label<S1: Into<String>, S2: Into<String>>(mut self, name: S1, value: S2) -> Self { |
| self.common_opts = self.common_opts.const_label(name, value); |
| self |
| } |
| |
| /// `variable_labels` sets the variable labels. |
| pub fn variable_labels(mut self, variable_labels: Vec<String>) -> Self { |
| self.common_opts = self.common_opts.variable_labels(variable_labels); |
| self |
| } |
| |
| /// `variable_label` adds a variable label. |
| pub fn variable_label<S: Into<String>>(mut self, name: S) -> Self { |
| self.common_opts = self.common_opts.variable_label(name); |
| self |
| } |
| |
| /// `fq_name` returns the fq_name. |
| pub fn fq_name(&self) -> String { |
| self.common_opts.fq_name() |
| } |
| |
| /// `buckets` set the buckets. |
| pub fn buckets(mut self, buckets: Vec<f64>) -> Self { |
| self.buckets = buckets; |
| self |
| } |
| } |
| |
| impl Describer for HistogramOpts { |
| fn describe(&self) -> Result<Desc> { |
| self.common_opts.describe() |
| } |
| } |
| |
| impl From<Opts> for HistogramOpts { |
| fn from(opts: Opts) -> HistogramOpts { |
| HistogramOpts { |
| common_opts: opts, |
| buckets: Vec::from(DEFAULT_BUCKETS as &'static [f64]), |
| } |
| } |
| } |
| |
| /// Representation of a hot or cold shard. |
| /// |
| /// See [`HistogramCore`] for details. |
| #[derive(Debug)] |
| struct Shard { |
| sum: AtomicF64, |
| count: AtomicU64, |
| buckets: Vec<AtomicU64>, |
| } |
| |
| impl Shard { |
| fn new(num_buckets: usize) -> Self { |
| let mut buckets = Vec::new(); |
| for _ in 0..num_buckets { |
| buckets.push(AtomicU64::new(0)); |
| } |
| |
| Shard { |
| sum: AtomicF64::new(0.0), |
| count: AtomicU64::new(0), |
| buckets, |
| } |
| } |
| } |
| |
| /// Index into an array of [`Shard`]s. |
| /// |
| /// Used in conjunction with [`ShardAndCount`] below. |
| #[derive(Debug, Clone, Copy)] |
| enum ShardIndex { |
| /// First index. Corresponds to 0. |
| First, |
| /// Second index. Corresponds to 1. |
| Second, |
| } |
| |
| impl ShardIndex { |
| /// Inverse the given [`ShardIndex`]. |
| fn inverse(self) -> ShardIndex { |
| match self { |
| ShardIndex::First => ShardIndex::Second, |
| ShardIndex::Second => ShardIndex::First, |
| } |
| } |
| } |
| |
| impl From<u64> for ShardIndex { |
| fn from(index: u64) -> Self { |
| match index { |
| 0 => ShardIndex::First, |
| 1 => ShardIndex::Second, |
| _ => panic!( |
| "Invalid shard index {:?}. A histogram only has two shards.", |
| index |
| ), |
| } |
| } |
| } |
| |
| impl From<ShardIndex> for usize { |
| fn from(index: ShardIndex) -> Self { |
| match index { |
| ShardIndex::First => 0, |
| ShardIndex::Second => 1, |
| } |
| } |
| } |
| |
| /// An atomic u64 with the most significant used as a [`ShardIndex`] and the |
| /// remaining 63 bits used to count [`Histogram`] observations. |
| #[derive(Debug)] |
| struct ShardAndCount { |
| inner: StdAtomicU64, |
| } |
| |
| impl ShardAndCount { |
| /// Return a new [`ShardAndCount`] with both the most significant bit |
| /// i.e. the `ShardIndex` and the remaining 63 bit i.e. the observation |
| /// count set to 0. |
| fn new() -> Self { |
| ShardAndCount { |
| inner: StdAtomicU64::new(0), |
| } |
| } |
| |
| /// Flip the most significant bit i.e. the [`ShardIndex`] leaving the |
| /// remaining 63 bits unchanged. |
| fn flip(&self, ordering: Ordering) -> (ShardIndex, u64) { |
| let n = self.inner.fetch_add(1 << 63, ordering); |
| |
| ShardAndCount::split_shard_index_and_count(n) |
| } |
| |
| /// Get the most significant bit i.e. the [`ShardIndex`] as well as the |
| /// remaining 63 bits i.e. the observation count. |
| fn get(&self) -> (ShardIndex, u64) { |
| let n = self.inner.load(Ordering::Relaxed); |
| |
| ShardAndCount::split_shard_index_and_count(n) |
| } |
| |
| /// Increment the observation count leaving the most significant bit i.e. |
| /// the [`ShardIndex`] untouched. |
| fn inc_by(&self, delta: u64, ordering: Ordering) -> (ShardIndex, u64) { |
| let n = self.inner.fetch_add(delta, ordering); |
| |
| ShardAndCount::split_shard_index_and_count(n) |
| } |
| |
| /// Increment the observation count by one leaving the most significant bit |
| /// i.e. the [`ShardIndex`] untouched. |
| fn inc(&self, ordering: Ordering) -> (ShardIndex, u64) { |
| self.inc_by(1, ordering) |
| } |
| |
| fn split_shard_index_and_count(n: u64) -> (ShardIndex, u64) { |
| let shard = n >> 63; |
| let count = n & ((1 << 63) - 1); |
| |
| (shard.into(), count) |
| } |
| } |
| |
| /// Core datastructure of a Prometheus histogram |
| /// |
| /// # Atomicity across collects |
| /// |
| /// A histogram supports two main execution paths: |
| /// |
| /// 1. `observe` which increases the overall observation counter, updates the |
| /// observation sum and increases a single bucket counter. |
| /// |
| /// 2. `proto` (aka. collecting the metric, from now on referred to as the |
| /// collect operation) which snapshots the state of the histogram and exposes it |
| /// as a Protobuf struct. |
| /// |
| /// If an observe and a collect operation interleave, the latter could be |
| /// exposing a snapshot of the histogram that does not uphold all histogram |
| /// invariants. For example for the invariant that the overall observation |
| /// counter should equal the sum of all bucket counters: Say that an `observe` |
| /// increases the overall counter but before updating a specific bucket counter |
| /// a collect operation snapshots the histogram. |
| /// |
| /// The below implementation of `HistogramCore` prevents such race conditions by |
| /// using two shards, one hot shard for `observe` operations to record their |
| /// observation and one cold shard for collect operations to collect a |
| /// consistent snapshot of the histogram. |
| /// |
| /// `observe` operations hit the hot shard and record their observation. Collect |
| /// operations switch hot and cold, wait for all `observe` calls to finish on |
| /// the previously hot now cold shard and then expose the consistent snapshot. |
| #[derive(Debug)] |
| pub struct HistogramCore { |
| desc: Desc, |
| label_pairs: Vec<proto::LabelPair>, |
| |
| /// Mutual exclusion to serialize collect operations. No two collect |
| /// operations should operate on this datastructure at the same time. (See |
| /// struct documentation for details.) `observe` operations can operate in |
| /// parallel without holding this lock. |
| collect_lock: Mutex<()>, |
| |
| /// An atomic u64 where the first bit determines the currently hot shard and |
| /// the remaining 63 bits determine the overall count. |
| shard_and_count: ShardAndCount, |
| /// The two shards where `shard_and_count` determines which one is the hot |
| /// and which one the cold at any given point in time. |
| shards: [Shard; 2], |
| |
| upper_bounds: Vec<f64>, |
| } |
| |
| impl HistogramCore { |
| pub fn new(opts: &HistogramOpts, label_values: &[&str]) -> Result<HistogramCore> { |
| let desc = opts.describe()?; |
| |
| for name in &desc.variable_labels { |
| check_bucket_label(name)?; |
| } |
| for pair in &desc.const_label_pairs { |
| check_bucket_label(pair.get_name())?; |
| } |
| |
| let label_pairs = make_label_pairs(&desc, label_values)?; |
| |
| let buckets = check_and_adjust_buckets(opts.buckets.clone())?; |
| |
| Ok(HistogramCore { |
| desc, |
| label_pairs, |
| |
| collect_lock: Mutex::new(()), |
| |
| shard_and_count: ShardAndCount::new(), |
| shards: [Shard::new(buckets.len()), Shard::new(buckets.len())], |
| |
| upper_bounds: buckets, |
| }) |
| } |
| |
| /// Record a given observation (f64) in the histogram. |
| // |
| // First increase the overall observation counter and thus learn which shard |
| // is the current hot shard. Subsequently on the hot shard update the |
| // corresponding bucket count, adjust the shard's sum and finally increase |
| // the shard's count. |
| pub fn observe(&self, v: f64) { |
| // The collect code path uses `self.shard_and_count` and |
| // `self.shards[x].count` to ensure not to collect data from a shard |
| // while observe calls are still operating on it. |
| // |
| // To ensure the above, this `inc` needs to use `Acquire` ordering to |
| // force anything below this line to stay below it. |
| let (shard_index, _count) = self.shard_and_count.inc(Ordering::Acquire); |
| |
| let shard: &Shard = &self.shards[usize::from(shard_index)]; |
| |
| // Try find the bucket. |
| let mut iter = self |
| .upper_bounds |
| .iter() |
| .enumerate() |
| .filter(|&(_, f)| v <= *f); |
| if let Some((i, _)) = iter.next() { |
| shard.buckets[i].inc_by(1); |
| } |
| |
| shard.sum.inc_by(v); |
| // Use `Release` ordering to ensure all operations above stay above. |
| shard.count.inc_by_with_ordering(1, Ordering::Release); |
| } |
| |
| /// Make a snapshot of the current histogram state exposed as a Protobuf |
| /// struct. |
| // |
| // Acquire the collect lock, switch the hot and the cold shard, wait for all |
| // remaining `observe` calls to finish on the previously hot now cold shard, |
| // snapshot the data, update the now hot shard and reset the cold shard. |
| pub fn proto(&self) -> proto::Histogram { |
| let collect_guard = self.collect_lock.lock().expect("Lock poisoned"); |
| |
| // `flip` needs to use AcqRel ordering to ensure the lock operation |
| // above stays above and the histogram operations (especially the shard |
| // resets) below stay below. |
| let (cold_shard_index, overall_count) = self.shard_and_count.flip(Ordering::AcqRel); |
| |
| let cold_shard = &self.shards[usize::from(cold_shard_index)]; |
| let hot_shard = &self.shards[usize::from(cold_shard_index.inverse())]; |
| |
| // Wait for all currently active `observe` calls on the now cold shard |
| // to finish. The above call to `flip` redirects all future `observe` |
| // calls to the other previously cold, now hot, shard. Thus once the |
| // cold shard counter equals the value of the global counter when the |
| // shards were flipped, all in-progress `observe` calls are done. With |
| // all of them done, the cold shard is now in a consistent state. |
| // |
| // `observe` uses `Release` ordering. `compare_exchange` needs to use |
| // `Acquire` ordering to ensure that (1) one sees all the previous |
| // `observe` stores to the counter and (2) to ensure the below shard |
| // modifications happen after this point, thus the shard is not modified |
| // by any `observe` operations. |
| while cold_shard |
| .count |
| .compare_exchange_weak( |
| overall_count, |
| // While at it, reset cold shard count on success. |
| 0, |
| Ordering::Acquire, |
| Ordering::Acquire, |
| ) |
| .is_err() |
| {} |
| |
| // Get cold shard sum and reset to 0. |
| // |
| // Use `Acquire` for load and `Release` for store to ensure not to |
| // interfere with previous or upcoming collect calls. |
| let cold_shard_sum = cold_shard.sum.swap(0.0, Ordering::AcqRel); |
| |
| let mut h = proto::Histogram::default(); |
| h.set_sample_sum(cold_shard_sum); |
| h.set_sample_count(overall_count); |
| |
| let mut cumulative_count = 0; |
| let mut buckets = Vec::with_capacity(self.upper_bounds.len()); |
| for (i, upper_bound) in self.upper_bounds.iter().enumerate() { |
| // Reset the cold shard and update the hot shard. |
| // |
| // Use `Acquire` for load and `Release` for store to ensure not to |
| // interfere with previous or upcoming collect calls. |
| let cold_bucket_count = cold_shard.buckets[i].swap(0, Ordering::AcqRel); |
| hot_shard.buckets[i].inc_by(cold_bucket_count); |
| |
| cumulative_count += cold_bucket_count; |
| let mut b = proto::Bucket::default(); |
| b.set_cumulative_count(cumulative_count); |
| b.set_upper_bound(*upper_bound); |
| buckets.push(b); |
| } |
| h.set_bucket(from_vec!(buckets)); |
| |
| // Update the hot shard. |
| hot_shard.count.inc_by(overall_count); |
| hot_shard.sum.inc_by(cold_shard_sum); |
| |
| drop(collect_guard); |
| |
| h |
| } |
| |
| fn sample_sum(&self) -> f64 { |
| // Make sure to not overlap with any collect calls, as they might flip |
| // the hot and cold shards. |
| let _guard = self.collect_lock.lock().expect("Lock poisoned"); |
| |
| let (shard_index, _count) = self.shard_and_count.get(); |
| self.shards[shard_index as usize].sum.get() |
| } |
| |
| fn sample_count(&self) -> u64 { |
| self.shard_and_count.get().1 |
| } |
| } |
| |
| // We have to wrap libc::timespec in order to implement std::fmt::Debug. |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| pub struct Timespec(libc::timespec); |
| |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| impl std::fmt::Debug for Timespec { |
| fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| write!( |
| f, |
| "Timespec {{ tv_sec: {}, tv_nsec: {} }}", |
| self.0.tv_sec, self.0.tv_nsec |
| ) |
| } |
| } |
| |
| #[derive(Debug)] |
| pub enum Instant { |
| Monotonic(StdInstant), |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| MonotonicCoarse(Timespec), |
| } |
| |
| impl Instant { |
| pub fn now() -> Instant { |
| Instant::Monotonic(StdInstant::now()) |
| } |
| |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| pub fn now_coarse() -> Instant { |
| Instant::MonotonicCoarse(get_time_coarse()) |
| } |
| |
| #[cfg(all(feature = "nightly", not(target_os = "linux")))] |
| pub fn now_coarse() -> Instant { |
| Instant::Monotonic(StdInstant::now()) |
| } |
| |
| pub fn elapsed(&self) -> Duration { |
| match self { |
| // We use `saturating_duration_since` to avoid panics caused by non-monotonic clocks. |
| Instant::Monotonic(i) => StdInstant::now().saturating_duration_since(*i), |
| |
| // It is different from `Instant::Monotonic`, the resolution here is millisecond. |
| // The processors in an SMP system do not start all at exactly the same time |
| // and therefore the timer registers are typically running at an offset. |
| // Use millisecond resolution for ignoring the error. |
| // See more: https://linux.die.net/man/2/clock_gettime |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| Instant::MonotonicCoarse(t) => { |
| let now = get_time_coarse(); |
| let now_ms = now.0.tv_sec * MILLIS_PER_SEC + now.0.tv_nsec / NANOS_PER_MILLI; |
| let t_ms = t.0.tv_sec * MILLIS_PER_SEC + t.0.tv_nsec / NANOS_PER_MILLI; |
| let dur = now_ms - t_ms; |
| if dur >= 0 { |
| Duration::from_millis(dur as u64) |
| } else { |
| Duration::from_millis(0) |
| } |
| } |
| } |
| } |
| |
| #[inline] |
| pub fn elapsed_sec(&self) -> f64 { |
| duration_to_seconds(self.elapsed()) |
| } |
| } |
| |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| use self::coarse::*; |
| |
| #[cfg(all(feature = "nightly", target_os = "linux"))] |
| mod coarse { |
| use crate::histogram::Timespec; |
| pub use libc::timespec; |
| use libc::{clock_gettime, CLOCK_MONOTONIC_COARSE}; |
| |
| pub const NANOS_PER_MILLI: i64 = 1_000_000; |
| pub const MILLIS_PER_SEC: i64 = 1_000; |
| |
| pub fn get_time_coarse() -> Timespec { |
| let mut t = Timespec(timespec { |
| tv_sec: 0, |
| tv_nsec: 0, |
| }); |
| assert_eq!( |
| unsafe { clock_gettime(CLOCK_MONOTONIC_COARSE, &mut t.0) }, |
| 0 |
| ); |
| t |
| } |
| } |
| |
| /// Timer to measure and record the duration of an event. |
| /// |
| /// This timer can be stopped and observed at most once, either automatically (when it |
| /// goes out of scope) or manually. |
| /// Alternatively, it can be manually stopped and discarded in order to not record its value. |
| #[must_use = "Timer should be kept in a variable otherwise it cannot observe duration"] |
| #[derive(Debug)] |
| pub struct HistogramTimer { |
| /// A histogram for automatic recording of observations. |
| histogram: Histogram, |
| /// Whether the timer has already been observed once. |
| observed: bool, |
| /// Starting instant for the timer. |
| start: Instant, |
| } |
| |
| impl HistogramTimer { |
| fn new(histogram: Histogram) -> Self { |
| Self { |
| histogram, |
| observed: false, |
| start: Instant::now(), |
| } |
| } |
| |
| #[cfg(feature = "nightly")] |
| fn new_coarse(histogram: Histogram) -> Self { |
| HistogramTimer { |
| histogram, |
| observed: false, |
| start: Instant::now_coarse(), |
| } |
| } |
| |
| /// Observe and record timer duration (in seconds). |
| /// |
| /// It observes the floating-point number of seconds elapsed since the timer |
| /// started, and it records that value to the attached histogram. |
| pub fn observe_duration(self) { |
| self.stop_and_record(); |
| } |
| |
| /// Observe, record and return timer duration (in seconds). |
| /// |
| /// It observes and returns a floating-point number for seconds elapsed since |
| /// the timer started, recording that value to the attached histogram. |
| pub fn stop_and_record(self) -> f64 { |
| let mut timer = self; |
| timer.observe(true) |
| } |
| |
| /// Observe and return timer duration (in seconds). |
| /// |
| /// It returns a floating-point number of seconds elapsed since the timer started, |
| /// without recording to any histogram. |
| pub fn stop_and_discard(self) -> f64 { |
| let mut timer = self; |
| timer.observe(false) |
| } |
| |
| fn observe(&mut self, record: bool) -> f64 { |
| let v = self.start.elapsed_sec(); |
| self.observed = true; |
| if record { |
| self.histogram.observe(v); |
| } |
| v |
| } |
| } |
| |
| impl Drop for HistogramTimer { |
| fn drop(&mut self) { |
| if !self.observed { |
| self.observe(true); |
| } |
| } |
| } |
| |
| /// A [`Metric`] counts individual observations from an event or sample stream |
| /// in configurable buckets. Similar to a [`Summary`](crate::proto::Summary), |
| /// it also provides a sum of observations and an observation count. |
| /// |
| /// On the Prometheus server, quantiles can be calculated from a [`Histogram`] using |
| /// the [`histogram_quantile`][1] function in the query language. |
| /// |
| /// Note that Histograms, in contrast to Summaries, can be aggregated with the |
| /// Prometheus query language (see [the prometheus documentation][2] for |
| /// detailed procedures). However, Histograms require the user to pre-define |
| /// suitable buckets, (see [`linear_buckets`] and [`exponential_buckets`] for |
| /// some helper provided here) and they are in general less accurate. The |
| /// Observe method of a [`Histogram`] has a very low performance overhead in |
| /// comparison with the Observe method of a Summary. |
| /// |
| /// [1]: https://prometheus.io/docs/prometheus/latest/querying/functions/#histogram_quantile |
| /// [2]: https://prometheus.io/docs/practices/histograms/ |
| #[derive(Clone, Debug)] |
| pub struct Histogram { |
| core: Arc<HistogramCore>, |
| } |
| |
| impl Histogram { |
| /// `with_opts` creates a [`Histogram`] with the `opts` options. |
| pub fn with_opts(opts: HistogramOpts) -> Result<Histogram> { |
| Histogram::with_opts_and_label_values(&opts, &[]) |
| } |
| |
| fn with_opts_and_label_values( |
| opts: &HistogramOpts, |
| label_values: &[&str], |
| ) -> Result<Histogram> { |
| let core = HistogramCore::new(opts, label_values)?; |
| |
| Ok(Histogram { |
| core: Arc::new(core), |
| }) |
| } |
| } |
| |
| impl Histogram { |
| /// Add a single observation to the [`Histogram`]. |
| pub fn observe(&self, v: f64) { |
| self.core.observe(v) |
| } |
| |
| /// Return a [`HistogramTimer`] to track a duration. |
| pub fn start_timer(&self) -> HistogramTimer { |
| HistogramTimer::new(self.clone()) |
| } |
| |
| /// Return a [`HistogramTimer`] to track a duration. |
| /// It is faster but less precise. |
| #[cfg(feature = "nightly")] |
| pub fn start_coarse_timer(&self) -> HistogramTimer { |
| HistogramTimer::new_coarse(self.clone()) |
| } |
| |
| /// Observe execution time of a closure, in second. |
| pub fn observe_closure_duration<F, T>(&self, f: F) -> T |
| where |
| F: FnOnce() -> T, |
| { |
| let instant = Instant::now(); |
| let res = f(); |
| let elapsed = instant.elapsed_sec(); |
| self.observe(elapsed); |
| res |
| } |
| |
| /// Observe execution time of a closure, in second. |
| #[cfg(feature = "nightly")] |
| pub fn observe_closure_duration_coarse<F, T>(&self, f: F) -> T |
| where |
| F: FnOnce() -> T, |
| { |
| let instant = Instant::now_coarse(); |
| let res = f(); |
| let elapsed = instant.elapsed_sec(); |
| self.observe(elapsed); |
| res |
| } |
| |
| /// Return a [`LocalHistogram`] for single thread usage. |
| pub fn local(&self) -> LocalHistogram { |
| LocalHistogram::new(self.clone()) |
| } |
| |
| /// Return accumulated sum of all samples. |
| pub fn get_sample_sum(&self) -> f64 { |
| self.core.sample_sum() |
| } |
| |
| /// Return count of all samples. |
| pub fn get_sample_count(&self) -> u64 { |
| self.core.sample_count() |
| } |
| } |
| |
| impl Metric for Histogram { |
| fn metric(&self) -> proto::Metric { |
| let mut m = proto::Metric::default(); |
| m.set_label(from_vec!(self.core.label_pairs.clone())); |
| |
| let h = self.core.proto(); |
| m.set_histogram(h); |
| |
| m |
| } |
| } |
| |
| impl Collector for Histogram { |
| fn desc(&self) -> Vec<&Desc> { |
| vec![&self.core.desc] |
| } |
| |
| fn collect(&self) -> Vec<proto::MetricFamily> { |
| let mut m = proto::MetricFamily::default(); |
| m.set_name(self.core.desc.fq_name.clone()); |
| m.set_help(self.core.desc.help.clone()); |
| m.set_field_type(proto::MetricType::HISTOGRAM); |
| m.set_metric(from_vec!(vec![self.metric()])); |
| |
| vec![m] |
| } |
| } |
| |
| #[derive(Clone, Debug)] |
| pub struct HistogramVecBuilder {} |
| |
| impl MetricVecBuilder for HistogramVecBuilder { |
| type M = Histogram; |
| type P = HistogramOpts; |
| |
| fn build(&self, opts: &HistogramOpts, vals: &[&str]) -> Result<Histogram> { |
| Histogram::with_opts_and_label_values(opts, vals) |
| } |
| } |
| |
| /// A [`Collector`] that bundles a set of Histograms that all share the |
| /// same [`Desc`], but have different values for their variable labels. This is used |
| /// if you want to count the same thing partitioned by various dimensions |
| /// (e.g. HTTP request latencies, partitioned by status code and method). |
| pub type HistogramVec = MetricVec<HistogramVecBuilder>; |
| |
| impl HistogramVec { |
| /// Create a new [`HistogramVec`] based on the provided |
| /// [`HistogramOpts`] and partitioned by the given label names. At least |
| /// one label name must be provided. |
| pub fn new(opts: HistogramOpts, label_names: &[&str]) -> Result<HistogramVec> { |
| let variable_names = label_names.iter().map(|s| (*s).to_owned()).collect(); |
| let opts = opts.variable_labels(variable_names); |
| let metric_vec = |
| MetricVec::create(proto::MetricType::HISTOGRAM, HistogramVecBuilder {}, opts)?; |
| |
| Ok(metric_vec as HistogramVec) |
| } |
| |
| /// Return a `LocalHistogramVec` for single thread usage. |
| pub fn local(&self) -> LocalHistogramVec { |
| let vec = self.clone(); |
| LocalHistogramVec::new(vec) |
| } |
| } |
| |
| /// Create `count` buckets, each `width` wide, where the lowest |
| /// bucket has an upper bound of `start`. The final +Inf bucket is not counted |
| /// and not included in the returned slice. The returned slice is meant to be |
| /// used for the Buckets field of [`HistogramOpts`]. |
| /// |
| /// The function returns an error if `count` is zero or `width` is zero or |
| /// negative. |
| pub fn linear_buckets(start: f64, width: f64, count: usize) -> Result<Vec<f64>> { |
| if count < 1 { |
| return Err(Error::Msg(format!( |
| "LinearBuckets needs a positive count, count: {}", |
| count |
| ))); |
| } |
| if width <= 0.0 { |
| return Err(Error::Msg(format!( |
| "LinearBuckets needs a width greater then 0, width: {}", |
| width |
| ))); |
| } |
| |
| let buckets: Vec<_> = (0..count) |
| .map(|step| start + width * (step as f64)) |
| .collect(); |
| |
| Ok(buckets) |
| } |
| |
| /// Create `count` buckets, where the lowest bucket has an |
| /// upper bound of `start` and each following bucket's upper bound is `factor` |
| /// times the previous bucket's upper bound. The final +Inf bucket is not counted |
| /// and not included in the returned slice. The returned slice is meant to be |
| /// used for the Buckets field of [`HistogramOpts`]. |
| /// |
| /// The function returns an error if `count` is zero, if `start` is zero or |
| /// negative, or if `factor` is less than or equal 1. |
| pub fn exponential_buckets(start: f64, factor: f64, count: usize) -> Result<Vec<f64>> { |
| if count < 1 { |
| return Err(Error::Msg(format!( |
| "exponential_buckets needs a positive count, count: {}", |
| count |
| ))); |
| } |
| if start <= 0.0 { |
| return Err(Error::Msg(format!( |
| "exponential_buckets needs a positive start value, \ |
| start: {}", |
| start |
| ))); |
| } |
| if factor <= 1.0 { |
| return Err(Error::Msg(format!( |
| "exponential_buckets needs a factor greater than 1, \ |
| factor: {}", |
| factor |
| ))); |
| } |
| |
| let mut next = start; |
| let mut buckets = Vec::with_capacity(count); |
| for _ in 0..count { |
| buckets.push(next); |
| next *= factor; |
| } |
| |
| Ok(buckets) |
| } |
| |
| /// `duration_to_seconds` converts Duration to seconds. |
| #[inline] |
| pub fn duration_to_seconds(d: Duration) -> f64 { |
| let nanos = f64::from(d.subsec_nanos()) / 1e9; |
| d.as_secs() as f64 + nanos |
| } |
| |
| #[derive(Clone, Debug)] |
| pub struct LocalHistogramCore { |
| histogram: Histogram, |
| counts: Vec<u64>, |
| count: u64, |
| sum: f64, |
| } |
| |
| /// An unsync [`Histogram`]. |
| #[derive(Debug)] |
| pub struct LocalHistogram { |
| core: RefCell<LocalHistogramCore>, |
| } |
| |
| impl Clone for LocalHistogram { |
| fn clone(&self) -> LocalHistogram { |
| let core = self.core.clone(); |
| let lh = LocalHistogram { core }; |
| lh.clear(); |
| lh |
| } |
| } |
| |
| /// An unsync [`HistogramTimer`]. |
| #[must_use = "Timer should be kept in a variable otherwise it cannot observe duration"] |
| #[derive(Debug)] |
| pub struct LocalHistogramTimer { |
| /// A local histogram for automatic recording of observations. |
| local: LocalHistogram, |
| /// Whether the timer has already been observed once. |
| observed: bool, |
| /// Starting instant for the timer. |
| start: Instant, |
| } |
| |
| impl LocalHistogramTimer { |
| fn new(histogram: LocalHistogram) -> Self { |
| Self { |
| local: histogram, |
| observed: false, |
| start: Instant::now(), |
| } |
| } |
| |
| #[cfg(feature = "nightly")] |
| fn new_coarse(histogram: LocalHistogram) -> Self { |
| Self { |
| local: histogram, |
| observed: false, |
| start: Instant::now_coarse(), |
| } |
| } |
| |
| /// Observe and record timer duration (in seconds). |
| /// |
| /// It observes the floating-point number of seconds elapsed since the timer |
| /// started, and it records that value to the attached histogram. |
| pub fn observe_duration(self) { |
| self.stop_and_record(); |
| } |
| |
| /// Observe, record and return timer duration (in seconds). |
| /// |
| /// It observes and returns a floating-point number for seconds elapsed since |
| /// the timer started, recording that value to the attached histogram. |
| pub fn stop_and_record(self) -> f64 { |
| let mut timer = self; |
| timer.observe(true) |
| } |
| |
| /// Observe and return timer duration (in seconds). |
| /// |
| /// It returns a floating-point number of seconds elapsed since the timer started, |
| /// without recording to any histogram. |
| pub fn stop_and_discard(self) -> f64 { |
| let mut timer = self; |
| timer.observe(false) |
| } |
| |
| fn observe(&mut self, record: bool) -> f64 { |
| let v = self.start.elapsed_sec(); |
| self.observed = true; |
| if record { |
| self.local.observe(v); |
| } |
| v |
| } |
| } |
| |
| impl Drop for LocalHistogramTimer { |
| fn drop(&mut self) { |
| if !self.observed { |
| self.observe(true); |
| } |
| } |
| } |
| |
| impl LocalHistogramCore { |
| fn new(histogram: Histogram) -> LocalHistogramCore { |
| let counts = vec![0; histogram.core.upper_bounds.len()]; |
| |
| LocalHistogramCore { |
| histogram, |
| counts, |
| count: 0, |
| sum: 0.0, |
| } |
| } |
| |
| 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 clear(&mut self) { |
| for v in &mut self.counts { |
| *v = 0 |
| } |
| |
| self.count = 0; |
| self.sum = 0.0; |
| } |
| |
| pub fn flush(&mut self) { |
| // No cached metric, return. |
| if self.count == 0 { |
| return; |
| } |
| |
| { |
| // The collect code path uses `self.shard_and_count` and |
| // `self.shards[x].count` to ensure not to collect data from a shard |
| // while observe calls are still operating on it. |
| // |
| // To ensure the above, this `inc` needs to use `Acquire` ordering |
| // to force anything below this line to stay below it. |
| let (shard_index, _count) = self |
| .histogram |
| .core |
| .shard_and_count |
| .inc_by(self.count, Ordering::Acquire); |
| let shard = &self.histogram.core.shards[shard_index as usize]; |
| |
| for (i, v) in self.counts.iter().enumerate() { |
| if *v > 0 { |
| shard.buckets[i].inc_by(*v); |
| } |
| } |
| |
| shard.sum.inc_by(self.sum); |
| // Use `Release` ordering to ensure all operations above stay above. |
| shard |
| .count |
| .inc_by_with_ordering(self.count, Ordering::Release); |
| } |
| |
| self.clear() |
| } |
| |
| fn sample_sum(&self) -> f64 { |
| self.sum |
| } |
| |
| fn sample_count(&self) -> u64 { |
| self.count |
| } |
| } |
| |
| impl LocalHistogram { |
| fn new(histogram: Histogram) -> LocalHistogram { |
| let core = LocalHistogramCore::new(histogram); |
| LocalHistogram { |
| core: RefCell::new(core), |
| } |
| } |
| |
| /// Add a single observation to the [`Histogram`]. |
| pub fn observe(&self, v: f64) { |
| self.core.borrow_mut().observe(v); |
| } |
| |
| /// Return a `LocalHistogramTimer` to track a duration. |
| pub fn start_timer(&self) -> LocalHistogramTimer { |
| LocalHistogramTimer::new(self.clone()) |
| } |
| |
| /// Return a `LocalHistogramTimer` to track a duration. |
| /// It is faster but less precise. |
| #[cfg(feature = "nightly")] |
| pub fn start_coarse_timer(&self) -> LocalHistogramTimer { |
| LocalHistogramTimer::new_coarse(self.clone()) |
| } |
| |
| /// Observe execution time of a closure, in second. |
| pub fn observe_closure_duration<F, T>(&self, f: F) -> T |
| where |
| F: FnOnce() -> T, |
| { |
| let instant = Instant::now(); |
| let res = f(); |
| let elapsed = instant.elapsed_sec(); |
| self.observe(elapsed); |
| res |
| } |
| |
| /// Observe execution time of a closure, in second. |
| #[cfg(feature = "nightly")] |
| pub fn observe_closure_duration_coarse<F, T>(&self, f: F) -> T |
| where |
| F: FnOnce() -> T, |
| { |
| let instant = Instant::now_coarse(); |
| let res = f(); |
| let elapsed = instant.elapsed_sec(); |
| self.observe(elapsed); |
| res |
| } |
| |
| /// Clear the local metric. |
| pub fn clear(&self) { |
| self.core.borrow_mut().clear(); |
| } |
| |
| /// Flush the local metrics to the [`Histogram`] metric. |
| pub fn flush(&self) { |
| self.core.borrow_mut().flush(); |
| } |
| |
| /// Return accumulated sum of local samples. |
| pub fn get_sample_sum(&self) -> f64 { |
| self.core.borrow().sample_sum() |
| } |
| |
| /// Return count of local samples. |
| pub fn get_sample_count(&self) -> u64 { |
| self.core.borrow().sample_count() |
| } |
| } |
| |
| impl LocalMetric for LocalHistogram { |
| /// Flush the local metrics to the [`Histogram`] metric. |
| fn flush(&self) { |
| LocalHistogram::flush(self); |
| } |
| } |
| |
| impl Drop for LocalHistogram { |
| fn drop(&mut self) { |
| self.flush() |
| } |
| } |
| |
| /// An unsync [`HistogramVec`]. |
| #[derive(Debug)] |
| pub struct LocalHistogramVec { |
| vec: HistogramVec, |
| local: HashMap<u64, LocalHistogram>, |
| } |
| |
| impl LocalHistogramVec { |
| fn new(vec: HistogramVec) -> LocalHistogramVec { |
| let local = HashMap::with_capacity(vec.v.children.read().len()); |
| LocalHistogramVec { vec, local } |
| } |
| |
| /// Get a [`LocalHistogram`] by label values. |
| /// See more [`MetricVec::with_label_values`]. |
| pub fn with_label_values<'a>(&'a mut self, vals: &[&str]) -> &'a LocalHistogram { |
| let hash = self.vec.v.hash_label_values(vals).unwrap(); |
| let vec = &self.vec; |
| self.local |
| .entry(hash) |
| .or_insert_with(|| vec.with_label_values(vals).local()) |
| } |
| |
| /// Remove a [`LocalHistogram`] by label values. |
| /// See more [`MetricVec::remove_label_values`]. |
| pub fn remove_label_values(&mut self, vals: &[&str]) -> Result<()> { |
| let hash = self.vec.v.hash_label_values(vals)?; |
| self.local.remove(&hash); |
| self.vec.v.delete_label_values(vals) |
| } |
| |
| /// Flush the local metrics to the [`HistogramVec`] metric. |
| pub fn flush(&self) { |
| for h in self.local.values() { |
| h.flush(); |
| } |
| } |
| } |
| |
| impl LocalMetric for LocalHistogramVec { |
| /// Flush the local metrics to the [`HistogramVec`] metric. |
| fn flush(&self) { |
| LocalHistogramVec::flush(self) |
| } |
| } |
| |
| impl Clone for LocalHistogramVec { |
| fn clone(&self) -> LocalHistogramVec { |
| LocalHistogramVec::new(self.vec.clone()) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use std::f64::{EPSILON, INFINITY}; |
| use std::thread; |
| use std::time::Duration; |
| |
| use super::*; |
| use crate::metrics::{Collector, Metric}; |
| |
| #[test] |
| fn test_histogram() { |
| let opts = HistogramOpts::new("test1", "test help") |
| .const_label("a", "1") |
| .const_label("b", "2"); |
| let histogram = Histogram::with_opts(opts).unwrap(); |
| histogram.observe(1.0); |
| |
| let timer = histogram.start_timer(); |
| thread::sleep(Duration::from_millis(100)); |
| timer.observe_duration(); |
| |
| let timer = histogram.start_timer(); |
| let handler = thread::spawn(move || { |
| let _timer = timer; |
| thread::sleep(Duration::from_millis(400)); |
| }); |
| assert!(handler.join().is_ok()); |
| |
| let mut mfs = histogram.collect(); |
| assert_eq!(mfs.len(), 1); |
| |
| let mf = mfs.pop().unwrap(); |
| let m = mf.get_metric().get(0).unwrap(); |
| assert_eq!(m.get_label().len(), 2); |
| let proto_histogram = m.get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), 3); |
| assert!(proto_histogram.get_sample_sum() >= 1.5); |
| assert_eq!(proto_histogram.get_bucket().len(), DEFAULT_BUCKETS.len()); |
| |
| let buckets = vec![1.0, 2.0, 3.0]; |
| let opts = HistogramOpts::new("test2", "test help").buckets(buckets.clone()); |
| let histogram = Histogram::with_opts(opts).unwrap(); |
| let mut mfs = histogram.collect(); |
| assert_eq!(mfs.len(), 1); |
| |
| let mf = mfs.pop().unwrap(); |
| let m = mf.get_metric().get(0).unwrap(); |
| assert_eq!(m.get_label().len(), 0); |
| let proto_histogram = m.get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), 0); |
| assert!((proto_histogram.get_sample_sum() - 0.0) < EPSILON); |
| assert_eq!(proto_histogram.get_bucket().len(), buckets.len()) |
| } |
| |
| #[test] |
| #[cfg(feature = "nightly")] |
| fn test_histogram_coarse_timer() { |
| let opts = HistogramOpts::new("test1", "test help"); |
| let histogram = Histogram::with_opts(opts).unwrap(); |
| |
| let timer = histogram.start_coarse_timer(); |
| thread::sleep(Duration::from_millis(100)); |
| timer.observe_duration(); |
| |
| let timer = histogram.start_coarse_timer(); |
| let handler = thread::spawn(move || { |
| let _timer = timer; |
| thread::sleep(Duration::from_millis(400)); |
| }); |
| assert!(handler.join().is_ok()); |
| |
| histogram.observe_closure_duration(|| { |
| thread::sleep(Duration::from_millis(400)); |
| }); |
| |
| let mut mfs = histogram.collect(); |
| assert_eq!(mfs.len(), 1); |
| |
| let mf = mfs.pop().unwrap(); |
| let m = mf.get_metric().get(0).unwrap(); |
| let proto_histogram = m.get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), 3); |
| assert!((proto_histogram.get_sample_sum() - 0.0) > EPSILON); |
| } |
| |
| #[test] |
| #[cfg(feature = "nightly")] |
| fn test_instant_on_smp() { |
| let zero = Duration::from_millis(0); |
| for i in 0..100_000 { |
| let now = Instant::now(); |
| let now_coarse = Instant::now_coarse(); |
| if i % 100 == 0 { |
| thread::yield_now(); |
| } |
| assert!(now.elapsed() >= zero); |
| assert!(now_coarse.elapsed() >= zero); |
| } |
| } |
| |
| #[test] |
| fn test_buckets_invalidation() { |
| let table = vec![ |
| (vec![], true, DEFAULT_BUCKETS.len()), |
| (vec![-2.0, -1.0, -0.5, 0.0, 0.5, 1.0, 2.0], true, 7), |
| (vec![-2.0, -1.0, -0.5, 10.0, 0.5, 1.0, 2.0], false, 7), |
| (vec![-2.0, -1.0, -0.5, 0.0, 0.5, 1.0, INFINITY], true, 6), |
| ]; |
| |
| for (buckets, is_ok, length) in table { |
| let got = check_and_adjust_buckets(buckets); |
| assert_eq!(got.is_ok(), is_ok); |
| if is_ok { |
| assert_eq!(got.unwrap().len(), length); |
| } |
| } |
| } |
| |
| #[test] |
| fn test_buckets_functions() { |
| let linear_table = vec![ |
| ( |
| -15.0, |
| 5.0, |
| 6, |
| true, |
| vec![-15.0, -10.0, -5.0, 0.0, 5.0, 10.0], |
| ), |
| (-15.0, 0.0, 6, false, vec![]), |
| (-15.0, 5.0, 0, false, vec![]), |
| ]; |
| |
| for (param1, param2, param3, is_ok, vec) in linear_table { |
| let got = linear_buckets(param1, param2, param3); |
| assert_eq!(got.is_ok(), is_ok); |
| if got.is_ok() { |
| assert_eq!(got.unwrap(), vec); |
| } |
| } |
| |
| let exponential_table = vec![ |
| (100.0, 1.2, 3, true, vec![100.0, 120.0, 144.0]), |
| (100.0, 0.5, 3, false, vec![]), |
| (100.0, 1.2, 0, false, vec![]), |
| ]; |
| |
| for (param1, param2, param3, is_ok, vec) in exponential_table { |
| let got = exponential_buckets(param1, param2, param3); |
| assert_eq!(got.is_ok(), is_ok); |
| if got.is_ok() { |
| assert_eq!(got.unwrap(), vec); |
| } |
| } |
| } |
| |
| #[test] |
| fn test_duration_to_seconds() { |
| let tbls = vec![(1000, 1.0), (1100, 1.1), (100_111, 100.111)]; |
| for (millis, seconds) in tbls { |
| let d = Duration::from_millis(millis); |
| let v = duration_to_seconds(d); |
| assert!((v - seconds).abs() < EPSILON); |
| } |
| } |
| |
| #[test] |
| fn test_histogram_vec_with_label_values() { |
| let vec = HistogramVec::new( |
| HistogramOpts::new("test_histogram_vec", "test histogram vec help"), |
| &["l1", "l2"], |
| ) |
| .unwrap(); |
| |
| assert!(vec.remove_label_values(&["v1", "v2"]).is_err()); |
| vec.with_label_values(&["v1", "v2"]).observe(1.0); |
| assert!(vec.remove_label_values(&["v1", "v2"]).is_ok()); |
| |
| assert!(vec.remove_label_values(&["v1"]).is_err()); |
| assert!(vec.remove_label_values(&["v1", "v3"]).is_err()); |
| } |
| |
| #[test] |
| fn test_histogram_vec_with_opts_buckets() { |
| let labels = ["l1", "l2"]; |
| let buckets = vec![1.0, 2.0, 3.0]; |
| let vec = HistogramVec::new( |
| HistogramOpts::new("test_histogram_vec", "test histogram vec help") |
| .buckets(buckets.clone()), |
| &labels, |
| ) |
| .unwrap(); |
| |
| let histogram = vec.with_label_values(&["v1", "v2"]); |
| histogram.observe(1.0); |
| |
| let m = histogram.metric(); |
| assert_eq!(m.get_label().len(), labels.len()); |
| |
| let proto_histogram = m.get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), 1); |
| assert!((proto_histogram.get_sample_sum() - 1.0) < EPSILON); |
| assert_eq!(proto_histogram.get_bucket().len(), buckets.len()) |
| } |
| |
| #[test] |
| fn test_histogram_local() { |
| let buckets = vec![1.0, 2.0, 3.0]; |
| let opts = HistogramOpts::new("test_histogram_local", "test histogram local help") |
| .buckets(buckets.clone()); |
| let histogram = Histogram::with_opts(opts).unwrap(); |
| let local = histogram.local(); |
| |
| let check = |count, sum| { |
| let m = histogram.metric(); |
| let proto_histogram = m.get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), count); |
| assert!((proto_histogram.get_sample_sum() - sum) < EPSILON); |
| }; |
| |
| local.observe(1.0); |
| local.observe(4.0); |
| check(0, 0.0); |
| |
| local.flush(); |
| check(2, 5.0); |
| |
| local.observe(2.0); |
| local.clear(); |
| check(2, 5.0); |
| |
| local.observe(2.0); |
| drop(local); |
| check(3, 7.0); |
| } |
| |
| #[test] |
| fn test_histogram_vec_local() { |
| let vec = HistogramVec::new( |
| HistogramOpts::new("test_histogram_vec_local", "test histogram vec help"), |
| &["l1", "l2"], |
| ) |
| .unwrap(); |
| let mut local_vec = vec.local(); |
| |
| vec.remove_label_values(&["v1", "v2"]).unwrap_err(); |
| local_vec.remove_label_values(&["v1", "v2"]).unwrap_err(); |
| |
| let check = |count, sum| { |
| let ms = vec.collect()[0].take_metric(); |
| let proto_histogram = ms[0].get_histogram(); |
| assert_eq!(proto_histogram.get_sample_count(), count); |
| assert!((proto_histogram.get_sample_sum() - sum) < EPSILON); |
| }; |
| |
| { |
| // Flush LocalHistogram |
| let h = local_vec.with_label_values(&["v1", "v2"]); |
| h.observe(1.0); |
| h.flush(); |
| check(1, 1.0); |
| } |
| |
| { |
| // Flush LocalHistogramVec |
| local_vec.with_label_values(&["v1", "v2"]).observe(4.0); |
| local_vec.flush(); |
| check(2, 5.0); |
| } |
| { |
| // Reset ["v1", "v2"] |
| local_vec.remove_label_values(&["v1", "v2"]).unwrap(); |
| |
| // Flush on drop |
| local_vec.with_label_values(&["v1", "v2"]).observe(2.0); |
| drop(local_vec); |
| check(1, 2.0); |
| } |
| } |
| |
| /// Ensure that when an observe and a collect operation interleave, the |
| /// latter does not expose a snapshot of the histogram that does not uphold |
| /// all histogram invariants. |
| #[test] |
| fn atomic_observe_across_collects() { |
| let done = Arc::new(std::sync::atomic::AtomicBool::default()); |
| let histogram = |
| Histogram::with_opts(HistogramOpts::new("test_name", "test help").buckets(vec![1.0])) |
| .unwrap(); |
| |
| let done_clone = done.clone(); |
| let histogram_clone = histogram.clone(); |
| let observing_thread = std::thread::spawn(move || loop { |
| if done_clone.load(std::sync::atomic::Ordering::Relaxed) { |
| break; |
| } |
| |
| for _ in 0..1_000_000 { |
| histogram_clone.observe(1.0); |
| } |
| }); |
| |
| let mut sample_count = 0; |
| let mut cumulative_count = 0; |
| let mut sample_sum = 0; |
| for _ in 0..1_000_000 { |
| let metric = &histogram.collect()[0].take_metric()[0]; |
| let proto = metric.get_histogram(); |
| |
| sample_count = proto.get_sample_count(); |
| sample_sum = proto.get_sample_sum() as u64; |
| // There is only one bucket thus the `[0]`. |
| cumulative_count = proto.get_bucket()[0].get_cumulative_count(); |
| |
| if sample_count != cumulative_count { |
| break; |
| } |
| |
| // Observation value is always `1.0` thus count and sum should |
| // always equal. The number of `observe` calls is limited to |
| // 1_000_000, thus the sum is limited to 1_000_000. A float 64 is |
| // able to represent the sum accurately up to 9_007_199_254_740_992. |
| if sample_count != sample_sum { |
| break; |
| } |
| } |
| |
| done.store(true, std::sync::atomic::Ordering::Relaxed); |
| observing_thread.join().unwrap(); |
| |
| if sample_count != cumulative_count { |
| panic!( |
| "Histogram invariant violated: For a histogram with a single \ |
| bucket observing values below the bucket's upper bound only \ |
| the histogram's count should always be equal to the buckets's \ |
| cumulative count, got {:?} and {:?} instead.", |
| sample_count, cumulative_count, |
| ); |
| } |
| |
| if sample_count != sample_sum { |
| panic!( |
| "Histogram invariant violated: For a histogram which is only \ |
| ever observing a value of `1.0` the sample count should equal \ |
| the sum, instead got: {:?} and {:?}", |
| sample_count, sample_sum, |
| ) |
| } |
| } |
| |
| #[test] |
| fn test_error_on_inconsistent_label_cardinality() { |
| let hist = Histogram::with_opts( |
| histogram_opts!( |
| "example_histogram", |
| "Used as an example", |
| vec![0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 5.0] |
| ) |
| .variable_label("example_variable"), |
| ); |
| |
| if let Err(Error::InconsistentCardinality { expect, got }) = hist { |
| assert_eq!(1, expect); |
| assert_eq!(0, got); |
| } else { |
| panic!("Expected InconsistentCardinality error.") |
| } |
| } |
| } |
