added histogram latency information to Hystrix dashboard stream

docs/root/operations/admin.rst

@@ -352,6 +352,7 @@ The fields are:
     In Envoy, service unavailable response will cause **outlier detection** - removing a node off the 
     load balancer pool, but requests are not rejected as a result. Therefore, this counter is always 
     set to '0'.
-  * Latency information is currently unavailable.
+  * Latency information represents data since last flush. 
+    Mean latency is currently not available.
 
 

source/common/stats/histogram_impl.cc

@@ -17,8 +17,8 @@ HistogramStatisticsImpl::HistogramStatisticsImpl(const histogram_t* histogram_pt
 }
 
 const std::vector<double>& HistogramStatisticsImpl::supportedQuantiles() const {
-  static const std::vector<double> supported_quantiles = {0,    0.25, 0.5,   0.75, 0.90,
-                                                          0.95, 0.99, 0.999, 1};
+  static const std::vector<double> supported_quantiles = {0,    0.25, 0.5,   0.75,  0.90,
+                                                          0.95, 0.99, 0.995, 0.999, 1};
   return supported_quantiles;
 }
 

source/extensions/stat_sinks/hystrix/BUILD

@@ -35,6 +35,7 @@ envoy_cc_library(
         "//include/envoy/stats:stats_interface",
         "//source/common/buffer:buffer_lib",
         "//source/common/common:logger_lib",
+        "//source/common/config:well_known_names",
         "//source/common/http:headers_lib",
     ],
 )

source/extensions/stat_sinks/hystrix/hystrix.cc

@@ -9,17 +9,19 @@
 
 #include "common/buffer/buffer_impl.h"
 #include "common/common/logger.h"
+#include "common/config/well_known_names.h"
 #include "common/http/headers.h"
 
 #include "absl/strings/str_cat.h"
+#include "absl/strings/str_split.h"
+#include "fmt/printf.h"
 
 namespace Envoy {
 namespace Extensions {
 namespace StatSinks {
 namespace Hystrix {
 
 const uint64_t HystrixSink::DEFAULT_NUM_BUCKETS;
-
 ClusterStatsCache::ClusterStatsCache(const std::string& cluster_name)
     : cluster_name_(cluster_name) {}
 
@@ -43,6 +45,18 @@ void ClusterStatsCache::printRollingWindow(absl::string_view name, RollingWindow
   out_str << std::endl;
 }
 
+void HystrixSink::addHistogramToStream(const QuantileLatencyMap& latency_map, absl::string_view key,
+                                       std::stringstream& ss) {
+  ss << ", \"" << key << "\": {";
+  bool is_first = true;
+  for (const std::pair<double, double>& element : latency_map) {
+    std::string quantile = fmt::sprintf("%g", element.first * 100);
+    HystrixSink::addDoubleToStream(quantile, element.second, ss, is_first);
+    is_first = false;
+  }
+  ss << "}";
+}
+
 // Add new value to rolling window, in place of oldest one.
 void HystrixSink::pushNewValue(RollingWindow& rolling_window, uint64_t value) {
   if (rolling_window.empty()) {
@@ -118,6 +132,11 @@ void HystrixSink::addIntToStream(absl::string_view key, uint64_t value, std::str
   addInfoToStream(key, std::to_string(value), info, is_first);
 }
 
+void HystrixSink::addDoubleToStream(absl::string_view key, double value, std::stringstream& info,
+                                    bool is_first) {
+  addInfoToStream(key, std::to_string(value), info, is_first);
+}
+
 void HystrixSink::addInfoToStream(absl::string_view key, absl::string_view value,
                                   std::stringstream& info, bool is_first) {
   if (!is_first) {
@@ -131,7 +150,7 @@ void HystrixSink::addHystrixCommand(ClusterStatsCache& cluster_stats_cache,
                                     absl::string_view cluster_name,
                                     uint64_t max_concurrent_requests, uint64_t reporting_hosts,
                                     std::chrono::milliseconds rolling_window_ms,
-                                    std::stringstream& ss) {
+                                    const QuantileLatencyMap& histogram, std::stringstream& ss) {
 
   std::time_t currentTime = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
 
@@ -161,7 +180,7 @@ void HystrixSink::addHystrixCommand(ClusterStatsCache& cluster_stats_cache,
   addIntToStream("rollingCountResponsesFromCache", 0, ss);
 
   // Envoy's "circuit breaker" has similar meaning to hystrix's isolation
-  // so we count upstream_rq_pending_overflow and present it as ss
+  // so we count upstream_rq_pending_overflow and present it as rollingCountSemaphoreRejected
   addIntToStream("rollingCountSemaphoreRejected", rejected, ss);
 
   // Hystrix's short circuit is not similar to Envoy's since it is triggered by 503 responses
@@ -173,12 +192,8 @@ void HystrixSink::addHystrixCommand(ClusterStatsCache& cluster_stats_cache,
   addIntToStream("rollingCountTimeout", timeouts, ss);
   addIntToStream("rollingCountBadRequests", 0, ss);
   addIntToStream("currentConcurrentExecutionCount", 0, ss);
-  addIntToStream("latencyExecute_mean", 0, ss);
-
-  // TODO trabetti : add histogram information once available by PR #2932
-  addInfoToStream(
-      "latencyExecute",
-      "{\"0\":0,\"25\":0,\"50\":0,\"75\":0,\"90\":0,\"95\":0,\"99\":0,\"99.5\":0,\"100\":0}", ss);
+  addStringToStream("latencyExecute_mean", "null", ss);
+  addHistogramToStream(histogram, "latencyExecute", ss);
   addIntToStream("propertyValue_circuitBreakerRequestVolumeThreshold", 0, ss);
   addIntToStream("propertyValue_circuitBreakerSleepWindowInMilliseconds", 0, ss);
   addIntToStream("propertyValue_circuitBreakerErrorThresholdPercentage", 0, ss);
@@ -230,10 +245,11 @@ void HystrixSink::addClusterStatsToStream(ClusterStatsCache& cluster_stats_cache
                                           uint64_t max_concurrent_requests,
                                           uint64_t reporting_hosts,
                                           std::chrono::milliseconds rolling_window_ms,
+                                          const QuantileLatencyMap& histogram,
                                           std::stringstream& ss) {
 
   addHystrixCommand(cluster_stats_cache, cluster_name, max_concurrent_requests, reporting_hosts,
-                    rolling_window_ms, ss);
+                    rolling_window_ms, histogram, ss);
   addHystrixThreadPool(cluster_name, max_concurrent_requests, reporting_hosts, rolling_window_ms,
                        ss);
 }
@@ -299,13 +315,43 @@ Http::Code HystrixSink::handlerHystrixEventStream(absl::string_view,
   return Http::Code::OK;
 }
 
-void HystrixSink::flush(Stats::Source&) {
+void HystrixSink::flush(Stats::Source& source) {
   if (callbacks_list_.empty()) {
     return;
   }
   incCounter();
   std::stringstream ss;
   Upstream::ClusterManager::ClusterInfoMap clusters = server_.clusterManager().clusters();
+
+  // Save a map of the relevant histograms per cluster in a convenient format.
+  std::unordered_map<std::string, QuantileLatencyMap> time_histograms;
+  for (const Stats::ParentHistogramSharedPtr histogram : source.cachedHistograms()) {
+    if (histogram->tagExtractedName() == "cluster.upstream_rq_time") {
+      // TODO(mrice32): add an Envoy utility function to look up and return a tag for a metric.
+      auto it = std::find_if(histogram->tags().begin(), histogram->tags().end(),
+                             [](const Stats::Tag& tag) {
+                               return (tag.name_ == Config::TagNames::get().CLUSTER_NAME);
+                             });
+
+      // Make sure we found the cluster name tag
+      ASSERT(it != histogram->tags().end()); // Can remove this??
+      auto it_bool_pair = time_histograms.emplace(std::make_pair(it->value_, QuantileLatencyMap()));
+      // Make sure histogram with this name was not already added
+      ASSERT(it_bool_pair.second);
+      QuantileLatencyMap& hist_map = it_bool_pair.first->second;
+
+      const std::vector<double>& supported_quantiles =
+          histogram->intervalStatistics().supportedQuantiles();
+      for (size_t i = 0; i < supported_quantiles.size(); ++i) {
+        // binary-search here is likely not worth it, as hystrix_quantiles has <10 elements.
+        if (std::find(hystrix_quantiles.begin(), hystrix_quantiles.end(), supported_quantiles[i]) !=
+            hystrix_quantiles.end()) {
+          hist_map[supported_quantiles[i]] = histogram->intervalStatistics().computedQuantiles()[i];
+        }
+      }
+    }
+  }
+
   for (auto& cluster : clusters) {
     Upstream::ClusterInfoConstSharedPtr cluster_info = cluster.second.get().info();
 
@@ -323,7 +369,7 @@ void HystrixSink::flush(Stats::Source&) {
         *cluster_stats_cache_ptr, cluster_info->name(),
         cluster_info->resourceManager(Upstream::ResourcePriority::Default).pendingRequests().max(),
         cluster_info->statsScope().gauge("membership_total").value(), server_.statsFlushInterval(),
-        ss);
+        time_histograms[cluster_info->name()], ss);
   }
 
   Buffer::OwnedImpl data;

source/extensions/stat_sinks/hystrix/hystrix.h

@@ -18,6 +18,10 @@ namespace Hystrix {
 typedef std::vector<uint64_t> RollingWindow;
 typedef std::map<const std::string, RollingWindow> RollingStatsMap;
 
+using QuantileLatencyMap = std::unordered_map<double, double>;
+static const std::vector<double> hystrix_quantiles = {0,    0.25, 0.5,   0.75, 0.90,
+                                                      0.95, 0.99, 0.995, 1};
+
 struct {
   const std::string AllowHeadersHystrix{"Accept, Cache-Control, X-Requested-With, Last-Event-ID"};
 } AccessControlAllowHeadersValue;
@@ -74,7 +78,8 @@ class HystrixSink : public Stats::Sink, public Logger::Loggable<Logger::Id::hyst
   void addClusterStatsToStream(ClusterStatsCache& cluster_stats_cache,
                                absl::string_view cluster_name, uint64_t max_concurrent_requests,
                                uint64_t reporting_hosts,
-                               std::chrono::milliseconds rolling_window_ms, std::stringstream& ss);
+                               std::chrono::milliseconds rolling_window_ms,
+                               const QuantileLatencyMap& histogram, std::stringstream& ss);
 
   /**
    * Calculate values needed to create the stream and write into the map.
@@ -96,33 +101,44 @@ class HystrixSink : public Stats::Sink, public Logger::Loggable<Logger::Id::hyst
    */
   uint64_t getRollingValue(RollingWindow rolling_window);
 
-private:
   /**
-   * Format the given key and absl::string_view value to "key"="value", and adding to the
+   * Format the given key and value to "key"=value, and adding to the stringstream.
+   */
+  static void addInfoToStream(absl::string_view key, absl::string_view value,
+                              std::stringstream& info, bool is_first = false);
+
+  /**
+   * Format the given key and double value to "key"=<string of uint64_t>, and adding to the
    * stringstream.
    */
-  void addStringToStream(absl::string_view key, absl::string_view value, std::stringstream& info,
-                         bool is_first = false);
+  static void addDoubleToStream(absl::string_view key, double value, std::stringstream& info,
+                                bool is_first);
 
   /**
-   * Format the given key and uint64_t value to "key"=<string of uint64_t>, and adding to the
+   * Format the given key and absl::string_view value to "key"="value", and adding to the
    * stringstream.
    */
-  void addIntToStream(absl::string_view key, uint64_t value, std::stringstream& info,
-                      bool is_first = false);
+  static void addStringToStream(absl::string_view key, absl::string_view value,
+                                std::stringstream& info, bool is_first = false);
 
   /**
-   * Format the given key and value to "key"=value, and adding to the stringstream.
+   * Format the given key and uint64_t value to "key"=<string of uint64_t>, and adding to the
+   * stringstream.
    */
-  void addInfoToStream(absl::string_view key, absl::string_view value, std::stringstream& info,
-                       bool is_first = false);
+  static void addIntToStream(absl::string_view key, uint64_t value, std::stringstream& info,
+                             bool is_first = false);
+
+  static void addHistogramToStream(const QuantileLatencyMap& latency_map, absl::string_view key,
+                                   std::stringstream& ss);
 
+private:
   /**
    * Generate HystrixCommand event stream.
    */
   void addHystrixCommand(ClusterStatsCache& cluster_stats_cache, absl::string_view cluster_name,
                          uint64_t max_concurrent_requests, uint64_t reporting_hosts,
-                         std::chrono::milliseconds rolling_window_ms, std::stringstream& ss);
+                         std::chrono::milliseconds rolling_window_ms,
+                         const QuantileLatencyMap& histogram, std::stringstream& ss);
 
   /**
    * Generate HystrixThreadPool event stream.

test/common/stats/thread_local_store_test.cc

@@ -610,9 +610,9 @@ TEST_F(HistogramTest, BasicHistogramSummaryValidate) {
 
   const std::string h1_expected_summary =
       "P0: 1, P25: 1.025, P50: 1.05, P75: 1.075, P90: 1.09, P95: 1.095, "
-      "P99: 1.099, P99.9: 1.0999, P100: 1.1";
-  const std::string h2_expected_summary =
-      "P0: 0, P25: 25, P50: 50, P75: 75, P90: 90, P95: 95, P99: 99, P99.9: 99.9, P100: 100";
+      "P99: 1.099, P99.5: 1.0995, P99.9: 1.0999, P100: 1.1";
+  const std::string h2_expected_summary = "P0: 0, P25: 25, P50: 50, P75: 75, P90: 90, P95: 95, "
+                                          "P99: 99, P99.5: 99.5, P99.9: 99.9, P100: 100";
 
   for (size_t i = 0; i < 100; ++i) {
     expectCallAndAccumulate(h2, i);
@@ -639,8 +639,8 @@ TEST_F(HistogramTest, BasicHistogramMergeSummary) {
   }
   EXPECT_EQ(1, validateMerge());
 
-  const std::string expected_summary =
-      "P0: 0, P25: 25, P50: 50, P75: 75, P90: 90, P95: 95, P99: 99, P99.9: 99.9, P100: 100";
+  const std::string expected_summary = "P0: 0, P25: 25, P50: 50, P75: 75, P90: 90, P95: 95, P99: "
+                                       "99, P99.5: 99.5, P99.9: 99.9, P100: 100";
 
   NameHistogramMap name_histogram_map = makeHistogramMap(store_->histograms());
   EXPECT_EQ(expected_summary, name_histogram_map["h1"]->cumulativeStatistics().summary());

test/extensions/stats_sinks/hystrix/hystrix_test.cc

@@ -9,13 +9,16 @@
 #include "test/mocks/upstream/mocks.h"
 
 #include "absl/strings/str_split.h"
+#include "circllhist.h"
+#include "fmt/printf.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
 using testing::InSequence;
 using testing::Invoke;
 using testing::NiceMock;
 using testing::Return;
+using testing::ReturnPointee;
 using testing::_;
 
 namespace Envoy {
@@ -206,6 +209,14 @@ class HystrixSinkTest : public testing::Test {
     return cluster_message_map;
   }
 
+  histogram_t* makeHistogram(const std::vector<uint64_t>& values) {
+    histogram_t* histogram = hist_alloc();
+    for (uint64_t value : values) {
+      hist_insert_intscale(histogram, value, 0, 1);
+    }
+    return histogram;
+  }
+
   TestRandomGenerator rand_;
   uint64_t window_size_ = rand_.random() % 10 + 5; // Arbitrary reasonable number.
   const std::string cluster1_name_{"test_cluster1"};
@@ -428,6 +439,61 @@ TEST_F(HystrixSinkTest, AddAndRemoveClusters) {
   // Check that old values of test_cluster2 were deleted.
   validateResults(cluster_message_map[cluster2_name_], 0, 0, 0, 0, 0, window_size_);
 }
+
+TEST_F(HystrixSinkTest, HistogramTest) {
+  InSequence s;
+  std::vector<Stats::ParentHistogramSharedPtr> stored_histograms;
+
+  // Create histogram for the Hystrix sink to read.
+  auto histogram = std::make_shared<NiceMock<Stats::MockParentHistogram>>();
+  histogram->name_ = "cluster." + cluster1_name_ + ".upstream_rq_time";
+  const std::string tag_extracted_name = "cluster.upstream_rq_time";
+  ON_CALL(*histogram, tagExtractedName())
+      .WillByDefault(testing::ReturnRefOfCopy(tag_extracted_name));
+  std::vector<Stats::Tag> tags;
+  Stats::Tag tag = {
+      Config::TagNames::get().CLUSTER_NAME, // name_
+      cluster1_name_                        // value_
+  };
+  tags.emplace_back(tag);
+  ON_CALL(*histogram, tags()).WillByDefault(testing::ReturnRef(tags));
+
+  histogram->used_ = true;
+
+  // Init with data such that the quantile value is equal to the quantile.
+  std::vector<uint64_t> h1_interval_values;
+  for (size_t i = 0; i < 100; ++i) {
+    h1_interval_values.push_back(i);
+  }
+
+  histogram_t* hist1_interval = makeHistogram(h1_interval_values);
+  Stats::HistogramStatisticsImpl h1_interval_statistics(hist1_interval);
+  ON_CALL(*histogram, intervalStatistics())
+      .WillByDefault(testing::ReturnRef(h1_interval_statistics));
+  stored_histograms.push_back(histogram);
+
+  ON_CALL(source_, cachedHistograms()).WillByDefault(ReturnPointee(&stored_histograms));
+
+  Buffer::OwnedImpl buffer = createClusterAndCallbacks();
+  // Register callback to sink.
+  sink_->registerConnection(&callbacks_);
+  sink_->flush(source_);
+
+  std::unordered_map<std::string, std::string> cluster_message_map =
+      buildClusterMap(buffer.toString());
+
+  Json::ObjectSharedPtr latency = Json::Factory::loadFromString(cluster_message_map[cluster1_name_])
+                                      ->getObject("latencyExecute");
+
+  // Data was added such that the value equals the quantile:
+  // "latencyExecute": {"99.5": 99.500000, "95": 95.000000, "90": 90.000000, "100": 100.000000, "0":
+  // 0.000000, "25": 25.000000, "99": 99.000000, "50": 50.000000, "75": 75.000000}.
+  for (auto it = begin(hystrix_quantiles); it != end(hystrix_quantiles); ++it) {
+    EXPECT_EQ(*it * 100, latency->getDouble(fmt::sprintf("%g", (*it) * 100)));
+  }
+
+  hist_free(hist1_interval);
+}
 } // namespace Hystrix
 } // namespace StatSinks
 } // namespace Extensions