support for hystrix event stream output

api/envoy/config/metrics/v2/stats.proto

@@ -184,6 +184,12 @@ message StatsdSink {
   string prefix = 3;
 }
 
+// Stats configuration proto schema for built-in *envoy.hystrix* sink.
+// The sink emits stats in SSE format for use with Hystrix dashboard
+message HystrixSink {
+  int64 num_of_buckets = 1;
+}
+
 // Stats configuration proto schema for built-in *envoy.dog_statsd* sink.
 // The sink emits stats with `DogStatsD <https://docs.datadoghq.com/guides/dogstatsd/>`_
 // compatible tags. Tags are configurable via :ref:`StatsConfig

include/envoy/http/header_map.h

@@ -285,6 +285,7 @@ class HeaderEntry {
   HEADER_FUNC(KeepAlive)                                                                           \
   HEADER_FUNC(LastModified)                                                                        \
   HEADER_FUNC(Method)                                                                              \
+  HEADER_FUNC(NoChunks)                                                                            \
   HEADER_FUNC(Origin)                                                                              \
   HEADER_FUNC(OtSpanContext)                                                                       \
   HEADER_FUNC(Path)                                                                                \

include/envoy/server/BUILD

@@ -25,6 +25,7 @@ envoy_cc_library(
         ":config_tracker_interface",
         "//include/envoy/buffer:buffer_interface",
         "//include/envoy/http:codes_interface",
+        "//include/envoy/http:filter_interface",
         "//include/envoy/network:listen_socket_interface",
     ],
 )

include/envoy/server/admin.h

@@ -6,6 +6,7 @@
 #include "envoy/buffer/buffer.h"
 #include "envoy/common/pure.h"
 #include "envoy/http/codes.h"
+#include "envoy/http/filter.h"
 #include "envoy/http/header_map.h"
 #include "envoy/network/listen_socket.h"
 #include "envoy/server/config_tracker.h"
@@ -15,6 +16,7 @@
 namespace Envoy {
 namespace Server {
 
+class AdminFilter;
 /**
  * This macro is used to add handlers to the Admin HTTP Endpoint. It builds
  * a callback that executes X when the specified admin handler is hit. This macro can be
@@ -23,8 +25,8 @@ namespace Server {
  */
 #define MAKE_ADMIN_HANDLER(X)                                                                      \
   [this](absl::string_view path_and_query, Http::HeaderMap& response_headers,                      \
-         Buffer::Instance& data) -> Http::Code {                                                   \
-    return X(path_and_query, response_headers, data);                                              \
+         Buffer::Instance& data, Server::AdminFilter& admin_filter) -> Http::Code {                \
+    return X(path_and_query, response_headers, data, admin_filter);                                \
   }
 
 /**
@@ -43,7 +45,9 @@ class Admin {
    * @return Http::Code the response code.
    */
   typedef std::function<Http::Code(absl::string_view path_and_query,
-                                   Http::HeaderMap& response_headers, Buffer::Instance& response)>
+                                   Http::HeaderMap& response_headers, Buffer::Instance& response,
+                                   AdminFilter& admin_filter)>
+
       HandlerCb;
 
   /**

include/envoy/server/instance.h

@@ -179,6 +179,11 @@ class Instance {
    * @return information about the local environment the server is running in.
    */
   virtual const LocalInfo::LocalInfo& localInfo() PURE;
+
+  /**
+   * @return the flush interval of stats sinks.
+   */
+  virtual std::chrono::milliseconds statsFlushInterval() PURE;
 };
 
 } // namespace Server

source/common/common/logger.h

@@ -32,6 +32,7 @@ namespace Logger {
   FUNCTION(hc)                   \
   FUNCTION(http)                 \
   FUNCTION(http2)                \
+  FUNCTION(hystrix)              \
   FUNCTION(lua)                  \
   FUNCTION(main)                 \
   FUNCTION(mongo)                \

source/common/http/headers.h

@@ -70,6 +70,7 @@ class HeaderValues {
   const LowerCaseString LastModified{"last-modified"};
   const LowerCaseString Location{"location"};
   const LowerCaseString Method{":method"};
+  const LowerCaseString NoChunks{":no-chunks"};
   const LowerCaseString Origin{"origin"};
   const LowerCaseString OtSpanContext{"x-ot-span-context"};
   const LowerCaseString Path{":path"};
@@ -104,12 +105,14 @@ class HeaderValues {
   } UpgradeValues;
 
   struct {
+    const std::string NoCache{"no-cache"};
     const std::string NoCacheMaxAge0{"no-cache, max-age=0"};
     const std::string NoTransform{"no-transform"};
   } CacheControlValues;
 
   struct {
     const std::string Text{"text/plain"};
+    const std::string TextEventStream{"text/event-stream"};
     const std::string TextUtf8{"text/plain; charset=UTF-8"}; // TODO(jmarantz): fold this into Text
     const std::string Html{"text/html; charset=UTF-8"};
     const std::string Grpc{"application/grpc"};
@@ -208,6 +211,15 @@ class HeaderValues {
     const std::string AcceptEncoding{"Accept-Encoding"};
     const std::string Wildcard{"*"};
   } VaryValues;
+
+  struct {
+    const std::string AccessControlAllowHeadersHystrix{
+        "Accept, Cache-Control, X-Requested-With, Last-Event-ID"};
+  } AccessControlAllowHeadersValue;
+
+  struct {
+    const std::string All{"*"};
+  } AccessControlAllowOriginValue;
 };
 
 typedef ConstSingleton<HeaderValues> Headers;

source/common/http/http1/codec_impl.cc

@@ -44,6 +44,7 @@ void StreamEncoderImpl::encode100ContinueHeaders(const HeaderMap& headers) {
 
 void StreamEncoderImpl::encodeHeaders(const HeaderMap& headers, bool end_stream) {
   bool saw_content_length = false;
+  bool no_chunks = false;
   headers.iterate(
       [](const HeaderEntry& header, void* context) -> HeaderMap::Iterate {
         const char* key_to_use = header.key().c_str();
@@ -69,12 +70,19 @@ void StreamEncoderImpl::encodeHeaders(const HeaderMap& headers, bool end_stream)
     saw_content_length = true;
   }
 
+  // for streaming (e.g. SSE stream sent to hystrix dashboard), we do not want
+  // chunk transfer encoding but we don't have a content-length so we pass "envoy only"
+  // header to avoid adding chunks
+  if (headers.NoChunks()) {
+    no_chunks = true;
+  }
+
   ASSERT(!headers.TransferEncoding());
 
   // Assume we are chunk encoding unless we are passed a content length or this is a header only
   // response. Upper layers generally should strip transfer-encoding since it only applies to
   // HTTP/1.1. The codec will infer it based on the type of response.
-  if (saw_content_length) {
+  if (saw_content_length || no_chunks) {
     chunk_encoding_ = false;
   } else {
     if (processing_100_continue_) {
@@ -107,8 +115,8 @@ void StreamEncoderImpl::encodeHeaders(const HeaderMap& headers, bool end_stream)
 }
 
 void StreamEncoderImpl::encodeData(Buffer::Instance& data, bool end_stream) {
-  // end_stream may be indicated with a zero length data buffer. If that is the case, so not
-  // atually write the zero length buffer out.
+  // end_stream may be indicated with a zero length data buffer. If that is the case, do not
+  // actually write the zero length buffer out.
   if (data.length() > 0) {
     if (chunk_encoding_) {
       connection_.buffer().add(fmt::format("{:x}\r\n", data.length()));

source/extensions/extensions_build_config.bzl

@@ -55,10 +55,10 @@ EXTENSIONS = {
     #
     # Stat sinks
     #
-
     "envoy.stat_sinks.dog_statsd":                      "//source/extensions/stat_sinks/dog_statsd:config",
     "envoy.stat_sinks.metrics_service":                 "//source/extensions/stat_sinks/metrics_service:config",
     "envoy.stat_sinks.statsd":                          "//source/extensions/stat_sinks/statsd:config",
+    "envoy.stat_sinks.hystrix":                         "//source/extensions/stat_sinks/hystrix:config",
 
     #
     # Tracers

source/extensions/stat_sinks/hystrix/BUILD

@@ -0,0 +1,38 @@
+licenses(["notice"])  # Apache 2
+# Stats sink for the basic version of the hystrix protocol (https://github.com/b/hystrix_spec).
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        "//include/envoy/registry",
+        "//source/common/network:address_lib",
+        "//source/common/network:resolver_lib",
+        "//source/extensions/stat_sinks:well_known_names",
+        "//source/extensions/stat_sinks/hystrix:hystrix_lib",
+        "//source/server:configuration_lib",
+        "@envoy_api//envoy/config/metrics/v2:stats_cc",
+    ],
+)
+
+envoy_cc_library(
+    name = "hystrix_lib",
+    srcs = ["hystrix.cc"],
+    hdrs = ["hystrix.h"],
+    deps = [
+        "//include/envoy/server:instance_interface",
+        "//include/envoy/stats:stats_interface",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:logger_lib",
+        "//source/server/http:admin_lib",
+    ],
+)

source/extensions/stat_sinks/hystrix/config.cc

@@ -0,0 +1,43 @@
+#include "extensions/stat_sinks/hystrix/config.h"
+
+#include "envoy/config/metrics/v2/stats.pb.h"
+#include "envoy/config/metrics/v2/stats.pb.validate.h"
+#include "envoy/registry/registry.h"
+
+#include "common/network/resolver_impl.h"
+
+#include "extensions/stat_sinks/hystrix/hystrix.h"
+#include "extensions/stat_sinks/well_known_names.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace StatSinks {
+namespace Hystrix {
+
+Stats::SinkPtr HystrixSinkFactory::createStatsSink(const Protobuf::Message& config,
+                                                   Server::Instance& server) {
+  const auto& hystrix_sink =
+      MessageUtil::downcastAndValidate<const envoy::config::metrics::v2::HystrixSink&>(config);
+  if (hystrix_sink.num_of_buckets() == 0) { // if not set
+    return std::make_unique<Hystrix::HystrixSink>(server);
+  }
+  return std::make_unique<Hystrix::HystrixSink>(server, hystrix_sink.num_of_buckets());
+}
+
+ProtobufTypes::MessagePtr HystrixSinkFactory::createEmptyConfigProto() {
+  return std::unique_ptr<envoy::config::metrics::v2::HystrixSink>(
+      new envoy::config::metrics::v2::HystrixSink());
+}
+
+std::string HystrixSinkFactory::name() { return StatsSinkNames::get().HYSTRIX; }
+
+/**
+ * Static registration for the statsd sink factory. @see RegisterFactory.
+ */
+static Registry::RegisterFactory<HystrixSinkFactory, Server::Configuration::StatsSinkFactory>
+    register_;
+
+} // namespace Hystrix
+} // namespace StatSinks
+} // namespace Extensions
+} // namespace Envoy

source/extensions/stat_sinks/hystrix/config.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <string>
+
+#include "envoy/server/instance.h"
+
+#include "server/configuration_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace StatSinks {
+namespace Hystrix {
+
+class HystrixSinkFactory : Logger::Loggable<Logger::Id::config>,
+                           public Server::Configuration::StatsSinkFactory {
+public:
+  // StatsSinkFactory
+  Stats::SinkPtr createStatsSink(const Protobuf::Message& config,
+                                 Server::Instance& server) override;
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override;
+
+  std::string name() override;
+};
+
+} // namespace Hystrix
+} // namespace StatSinks
+} // namespace Extensions
+} // namespace Envoy