return histograms for time metrics

bin/src/command/requests.rs

@@ -255,7 +255,7 @@
 #[derive(Debug)]
 pub struct QueryClustersTask {
     pub client_token: Token,
     pub request_type: RequestType,
     pub gatherer: DefaultGatherer,
     main_process_response: Option<ResponseContent>,
 }
@@ -555,6 +555,11 @@
                     summed_cluster_metrics.append(&mut listed_cluster_metrics.clone());
                 }
             }
+            summed_proxy_metrics.sort();
+            summed_cluster_metrics.sort();
+            summed_proxy_metrics.dedup();
+            summed_cluster_metrics.dedup();
+
             return client.finish_ok_with_content(
                 ContentType::AvailableMetrics(AvailableMetrics {
                     proxy_metrics: summed_proxy_metrics,

command/src/command.proto

@@ -614,6 +614,7 @@ message BackendMetrics {
     map<string, FilteredMetrics> metrics = 2;
 }
 
+// A metric, in a "filtered" format, which means: sendable to outside programs.
 message FilteredMetrics {
     oneof inner {
         // increases or decrease depending on the state
@@ -624,6 +625,7 @@ message FilteredMetrics {
         uint64 time = 3;
         Percentiles percentiles = 4;
         FilteredTimeSerie time_serie = 5;
+        FilteredHistogram histogram = 6;
     }
 }
 
@@ -633,7 +635,6 @@ message FilteredTimeSerie {
     repeated uint32 last_hour = 3;
 }
 
-
 message Percentiles {
     required uint64 samples = 1;
     required uint64 p_50 = 2;
@@ -646,6 +647,20 @@ message Percentiles {
     required uint64 sum = 9;
 }
 
+// a histogram meant to be translated to prometheus
+message FilteredHistogram {
+    required uint64 sum = 1;
+    required uint64 count = 2;
+    repeated Bucket buckets = 3;
+}
+
+// a prometheus histogram bucket
+message Bucket {
+    required uint64 count = 1;
+    // upper range of the bucket (le = less or equal)
+    required uint64 le = 2;
+}
+
 message RequestCounts {
     map<string, int32> map = 1;
 }

command/src/proto/display.rs

@@ -25,6 +25,8 @@ use crate::{
     AsString,
 };
 
+use super::command::FilteredHistogram;
+
 impl Display for CertificateAndKey {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         let versions = self.versions.iter().fold(String::new(), |acc, tls_v| {
@@ -240,14 +242,14 @@ pub fn print_metrics(aggregated_metrics: &AggregatedMetrics) -> Result<(), Displ
     println!("\nMAIN PROCESS\n============");
     print_proxy_metrics(&aggregated_metrics.main);
 
-    if aggregated_metrics.proxying.len() != 0 {
+    if !aggregated_metrics.proxying.is_empty() {
         println!("\nPROXYING\n============");
         print_proxy_metrics(&aggregated_metrics.proxying);
     }
 
     // workers
     for (worker_id, worker) in aggregated_metrics.workers.iter() {
-        if worker.clusters.len() != 0 && worker.proxy.len() != 0 {
+        if !worker.clusters.is_empty() && !worker.proxy.is_empty() {
             println!("\nWorker {worker_id}\n=========");
             print_worker_metrics(worker)?;
         }
@@ -266,6 +268,7 @@ fn print_proxy_metrics(proxy_metrics: &BTreeMap<String, FilteredMetrics>) {
     let filtered = filter_metrics(proxy_metrics);
     print_gauges_and_counts(&filtered);
     print_percentiles(&filtered);
+    print_histograms(&filtered);
 }
 
 fn print_worker_metrics(worker_metrics: &WorkerMetrics) -> Result<(), DisplayError> {
@@ -282,6 +285,7 @@ fn print_cluster_metrics(cluster_metrics: &BTreeMap<String, ClusterMetrics>) {
         let filtered = filter_metrics(&cluster_metrics_data.cluster);
         print_gauges_and_counts(&filtered);
         print_percentiles(&filtered);
+        print_histograms(&filtered);
 
         // backend_id -> (metric_name -> value )
         let mut backend_metric_acc: HashMap<String, BTreeMap<String, FilteredMetrics>> =
@@ -303,6 +307,7 @@ fn print_cluster_metrics(cluster_metrics: &BTreeMap<String, ClusterMetrics>) {
             let filtered = filter_metrics(&metrics);
             print_gauges_and_counts(&filtered);
             print_percentiles(&filtered);
+            print_histograms(&filtered);
         }
     }
 }
@@ -421,6 +426,59 @@ fn print_percentiles(filtered_metrics: &BTreeMap<String, FilteredMetrics>) {
     percentile_table.printstd();
 }
 
+fn print_histograms(filtered_metrics: &BTreeMap<String, FilteredMetrics>) {
+    let histograms: BTreeMap<String, FilteredHistogram> = filtered_metrics
+        .iter()
+        .filter_map(|(name, metric)| match metric.inner.clone() {
+            Some(filtered_metrics::Inner::Histogram(hist)) => Some((name.to_owned(), hist)),
+            _ => None,
+        })
+        .collect();
+
+    let mut histogram_titles: Vec<String> = histograms.keys().map(ToOwned::to_owned).collect();
+    histogram_titles.sort();
+    if histogram_titles.is_empty() {
+        return;
+    }
+
+    let mut histogram_table = Table::new();
+    histogram_table.set_format(*prettytable::format::consts::FORMAT_BOX_CHARS);
+
+    let mut first_row = Row::new(vec![cell!("Histograms (ms)"), cell!("sum"), cell!("count")]);
+
+    let biggest_hist_length = histograms
+        .values()
+        .map(|hist| hist.buckets.len())
+        .max()
+        .unwrap_or(0);
+
+    // 0, 1, 3, 7... in the upper row
+    for exponent in 0..biggest_hist_length {
+        first_row.add_cell(cell!(format!("{}", (1 << exponent) - 1)));
+    }
+    histogram_table.set_titles(first_row);
+
+    for title in histogram_titles {
+        if let Some(hist) = histograms.get(&title) {
+            let trimmed_name = title.strip_suffix("_histogram").unwrap_or_default();
+            let mut row = Row::new(vec![
+                cell!(trimmed_name),
+                cell!(hist.sum),
+                cell!(hist.count),
+            ]);
+            // display the count by bucket, not the incremented count
+            let mut last_bucket_count = 0;
+            for bucket in &hist.buckets {
+                row.add_cell(cell!(bucket.count - last_bucket_count));
+                last_bucket_count = bucket.count;
+            }
+            histogram_table.add_row(row);
+        }
+    }
+
+    histogram_table.printstd();
+}
+
 fn print_available_metrics(available_metrics: &AvailableMetrics) -> Result<(), DisplayError> {
     println!("Available metrics on the proxy level:");
     for metric_name in &available_metrics.proxy_metrics {

command/src/proto/mod.rs

@@ -1,6 +1,9 @@
 use std::collections::BTreeMap;
 
-use command::{filtered_metrics::Inner, AggregatedMetrics, ClusterMetrics, FilteredMetrics};
+use command::{
+    filtered_metrics::Inner, AggregatedMetrics, BackendMetrics, Bucket, FilteredHistogram,
+    FilteredMetrics,
+};
 use prost::DecodeError;
 
 /// Contains all types received by and sent from Sōzu
@@ -53,63 +56,50 @@ impl AggregatedMetrics {
 
         for (_worker_id, worker) in workers {
             for (metric_name, new_value) in worker.proxy {
-                if !new_value.is_mergeable() {
-                    continue;
+                if new_value.is_mergeable() {
+                    self.proxying
+                        .entry(metric_name)
+                        .and_modify(|old_value| old_value.merge(&new_value))
+                        .or_insert(new_value);
                 }
-                self.proxying
-                    .entry(metric_name)
-                    .and_modify(|old_value| old_value.merge(&new_value))
-                    .or_insert(new_value);
             }
 
             for (cluster_id, mut cluster_metrics) in worker.clusters {
                 for (metric_name, new_value) in cluster_metrics.cluster {
-                    if !new_value.is_mergeable() {
-                        continue;
+                    if new_value.is_mergeable() {
+                        let cluster = self.clusters.entry(cluster_id.to_owned()).or_default();
+
+                        cluster
+                            .cluster
+                            .entry(metric_name)
+                            .and_modify(|old_value| old_value.merge(&new_value))
+                            .or_insert(new_value);
                     }
-                    self.clusters
-                        .entry(cluster_id.to_owned())
-                        .and_modify(|cluster| {
-                            cluster
-                                .cluster
-                                .entry(metric_name.clone())
-                                .and_modify(|old_value| old_value.merge(&new_value))
-                                .or_insert(new_value.clone());
-                        })
-                        .or_insert(ClusterMetrics {
-                            cluster: BTreeMap::from([(metric_name, new_value)]),
-                            backends: Vec::new(),
-                        });
                 }
 
                 for backend in cluster_metrics.backends.drain(..) {
-                    for (metric_name, new_value) in &backend.metrics {
-                        if !new_value.is_mergeable() {
-                            continue;
-                        }
-                        self.clusters
-                            .entry(cluster_id.to_owned())
-                            .and_modify(|cluster| {
-                                let found_backend = cluster
-                                    .backends
-                                    .iter_mut()
-                                    .find(|present| present.backend_id == backend.backend_id);
-
-                                let Some(existing_backend) = found_backend else {
-                                    cluster.backends.push(backend.clone());
-                                    return;
-                                };
+                    for (metric_name, new_value) in backend.metrics {
+                        if new_value.is_mergeable() {
+                            let cluster = self.clusters.entry(cluster_id.to_owned()).or_default();
+
+                            let found_backend = cluster
+                                .backends
+                                .iter_mut()
+                                .find(|present| &present.backend_id == &backend.backend_id);
 
+                            if let Some(existing_backend) = found_backend {
                                 let _ = existing_backend
                                     .metrics
-                                    .entry(metric_name.clone())
+                                    .entry(metric_name)
                                     .and_modify(|old_value| old_value.merge(&new_value))
-                                    .or_insert(new_value.to_owned());
-                            })
-                            .or_insert(ClusterMetrics {
-                                cluster: BTreeMap::new(),
-                                backends: vec![backend.clone()],
-                            });
+                                    .or_insert(new_value);
+                            } else {
+                                cluster.backends.push(BackendMetrics {
+                                    backend_id: backend.backend_id.clone(),
+                                    metrics: BTreeMap::from([(metric_name, new_value)]),
+                                });
+                            };
+                        }
                     }
                 }
             }
@@ -130,13 +120,39 @@ impl FilteredMetrics {
                     inner: Some(Inner::Count(a + b)),
                 };
             }
+            (Some(Inner::Histogram(a)), Some(Inner::Histogram(b))) => {
+                let longest_len = a.buckets.len().max(b.buckets.len());
+
+                let buckets = (0..longest_len)
+                    .map(|i| Bucket {
+                        le: (1 << i) - 1, // the bucket less-or-equal limits are normalized: 0, 1, 3, 7, 15, ...
+                        count: a
+                            .buckets
+                            .get(i)
+                            .and_then(|buck| Some(buck.count))
+                            .unwrap_or(0)
+                            + b.buckets
+                                .get(i)
+                                .and_then(|buck| Some(buck.count))
+                                .unwrap_or(0),
+                    })
+                    .collect();
+
+                *self = Self {
+                    inner: Some(Inner::Histogram(FilteredHistogram {
+                        count: a.count + b.count,
+                        sum: a.sum + b.sum,
+                        buckets,
+                    })),
+                };
+            }
             _ => {}
         }
     }
 
     fn is_mergeable(&self) -> bool {
         match &self.inner {
-            Some(Inner::Gauge(_)) | Some(Inner::Count(_)) => true,
+            Some(Inner::Gauge(_)) | Some(Inner::Count(_)) | Some(Inner::Histogram(_)) => true,
             // Inner::Time and Inner::Timeserie are never used in Sōzu
             Some(Inner::Time(_))
             | Some(Inner::Percentiles(_))