Rolling summary configuration (#444)

metrics-exporter-prometheus/src/builder.rs

@@ -5,6 +5,7 @@ use std::convert::TryFrom;
 use std::future::Future;
 #[cfg(feature = "http-listener")]
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::num::NonZeroU32;
 #[cfg(any(feature = "http-listener", feature = "push-gateway"))]
 use std::pin::Pin;
 use std::sync::RwLock;
@@ -96,6 +97,8 @@ pub struct PrometheusBuilder {
     #[cfg(feature = "http-listener")]
     allowed_addresses: Option<Vec<IpNet>>,
     quantiles: Vec<Quantile>,
+    bucket_duration: Option<Duration>,
+    bucket_count: Option<NonZeroU32>,
     buckets: Option<Vec<f64>>,
     bucket_overrides: Option<HashMap<Matcher, Vec<f64>>>,
     idle_timeout: Option<Duration>,
@@ -120,6 +123,8 @@ impl PrometheusBuilder {
             #[cfg(feature = "http-listener")]
             allowed_addresses: None,
             quantiles,
+            bucket_duration: None,
+            bucket_count: None,
             buckets: None,
             bucket_overrides: None,
             idle_timeout: None,
@@ -239,6 +244,30 @@ impl PrometheusBuilder {
         Ok(self)
     }
 
+    /// Sets the default bucket duration for rolling summaries
+    ///
+    /// Buckets will be cleared after this interval expires
+    ///
+    /// ## Errors
+    ///
+    /// If `value` less than 1 error will be thrown
+    pub fn set_bucket_duration(mut self, value: Duration) -> Result<Self, BuildError> {
+        if value.is_zero() {
+            return Err(BuildError::ZeroBucketDuration);
+        }
+
+        self.bucket_duration = Some(value);
+        Ok(self)
+    }
+
+    /// Sets the default bucket count for rolling summaries
+    ///
+    /// Count number buckets are created to store summary information
+    pub fn set_bucket_count(mut self, count: NonZeroU32) -> Self {
+        self.bucket_count = Some(count);
+        self
+    }
+
     /// Sets the buckets to use when rendering histograms.
     ///
     /// Buckets values represent the higher bound of each buckets.  If buckets are set, then all
@@ -536,7 +565,9 @@ impl PrometheusBuilder {
             distributions: RwLock::new(HashMap::new()),
             distribution_builder: DistributionBuilder::new(
                 self.quantiles,
+                self.bucket_duration,
                 self.buckets,
+                self.bucket_count,
                 self.bucket_overrides,
             ),
             descriptions: RwLock::new(HashMap::new()),

metrics-exporter-prometheus/src/common.rs

@@ -74,6 +74,10 @@ pub enum BuildError {
     /// Bucket bounds or quantiles were empty.
     #[error("bucket bounds/quantiles cannot be empty")]
     EmptyBucketsOrQuantiles,
+
+    /// Bucket duration cannot be zero
+    #[error("bucket durations cannot be set to zero")]
+    ZeroBucketDuration,
 }
 
 pub struct Snapshot {

metrics-exporter-prometheus/src/distribution.rs

@@ -8,6 +8,9 @@ use crate::common::Matcher;
 
 use metrics_util::{Histogram, Quantile, Summary};
 
+const DEFAULT_SUMMARY_BUCKET_COUNT: u32 = 3;
+const DEFAULT_SUMMARY_BUCKET_DURATION: u64 = 20;
+
 /// Distribution type.
 #[derive(Clone)]
 pub enum Distribution {
@@ -34,9 +37,12 @@ impl Distribution {
     }
 
     /// Creates a summary distribution.
-    pub fn new_summary(quantiles: Arc<Vec<Quantile>>) -> Distribution {
-        let summary = RollingSummary::default();
-        Distribution::Summary(summary, quantiles, 0.0)
+    pub fn new_summary(
+        quantiles: Arc<Vec<Quantile>>,
+        bucket_duration: Duration,
+        bucket_count: NonZeroU32,
+    ) -> Distribution {
+        Distribution::Summary(RollingSummary::new(bucket_count, bucket_duration), quantiles, 0.0)
     }
 
     /// Records the given `samples` in the current distribution.
@@ -60,19 +66,25 @@ impl Distribution {
 pub struct DistributionBuilder {
     quantiles: Arc<Vec<Quantile>>,
     buckets: Option<Vec<f64>>,
+    bucket_duration: Option<Duration>,
+    bucket_count: Option<NonZeroU32>,
     bucket_overrides: Option<Vec<(Matcher, Vec<f64>)>>,
 }
 
 impl DistributionBuilder {
     /// Creates a new instance of `DistributionBuilder`.
     pub fn new(
         quantiles: Vec<Quantile>,
+        bucket_duration: Option<Duration>,
         buckets: Option<Vec<f64>>,
+        bucket_count: Option<NonZeroU32>,
         bucket_overrides: Option<HashMap<Matcher, Vec<f64>>>,
     ) -> DistributionBuilder {
         DistributionBuilder {
             quantiles: Arc::new(quantiles),
+            bucket_duration,
             buckets,
+            bucket_count,
             bucket_overrides: bucket_overrides.map(|entries| {
                 let mut matchers = entries.into_iter().collect::<Vec<_>>();
                 matchers.sort_by(|a, b| a.0.cmp(&b.0));
@@ -95,7 +107,13 @@ impl DistributionBuilder {
             return Distribution::new_histogram(buckets);
         }
 
-        Distribution::new_summary(self.quantiles.clone())
+        let b_duration = self
+            .bucket_duration
+            .map_or(Duration::from_secs(DEFAULT_SUMMARY_BUCKET_DURATION), |d| d);
+        let b_count =
+            self.bucket_count.map_or(NonZeroU32::new(DEFAULT_SUMMARY_BUCKET_COUNT).unwrap(), |c| c);
+
+        Distribution::new_summary(self.quantiles.clone(), b_duration, b_count)
     }
 
     /// Returns the distribution type for the given metric key.
@@ -142,7 +160,10 @@ pub struct RollingSummary {
 
 impl Default for RollingSummary {
     fn default() -> Self {
-        RollingSummary::new(NonZeroU32::new(3).unwrap(), Duration::from_secs(20))
+        RollingSummary::new(
+            NonZeroU32::new(DEFAULT_SUMMARY_BUCKET_COUNT).unwrap(),
+            Duration::from_secs(20),
+        )
     }
 }