diff --git a/api/envoy/config/metrics/v2/stats.proto b/api/envoy/config/metrics/v2/stats.proto index 7aab95a43961c..ecd58fae2c77c 100644 --- a/api/envoy/config/metrics/v2/stats.proto +++ b/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 `_ // compatible tags. Tags are configurable via :ref:`StatsConfig diff --git a/include/envoy/http/header_map.h b/include/envoy/http/header_map.h index 999e990bfa2d0..29c3ed3e7d13b 100644 --- a/include/envoy/http/header_map.h +++ b/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) \ diff --git a/include/envoy/server/BUILD b/include/envoy/server/BUILD index b0bb59c5b6ab9..bd7dfda21ce3a 100644 --- a/include/envoy/server/BUILD +++ b/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", ], ) diff --git a/include/envoy/server/admin.h b/include/envoy/server/admin.h index 92302f9d664ab..3c44f85fa203f 100644 --- a/include/envoy/server/admin.h +++ b/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::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter& admin_filter)> + HandlerCb; /** diff --git a/include/envoy/server/instance.h b/include/envoy/server/instance.h index af7fd56e11f4e..139b59326e447 100644 --- a/include/envoy/server/instance.h +++ b/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 diff --git a/source/common/common/logger.h b/source/common/common/logger.h index 9d1b0b3661680..b6ce4d361d01b 100644 --- a/source/common/common/logger.h +++ b/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) \ diff --git a/source/common/http/headers.h b/source/common/http/headers.h index bb58c0c8418e2..430dd0274aa15 100644 --- a/source/common/http/headers.h +++ b/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 Headers; diff --git a/source/common/http/http1/codec_impl.cc b/source/common/http/http1/codec_impl.cc index 548eddd086c6b..d6754ee838d94 100644 --- a/source/common/http/http1/codec_impl.cc +++ b/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())); diff --git a/source/extensions/extensions_build_config.bzl b/source/extensions/extensions_build_config.bzl index acafc31c9aae7..3836b8ab8a238 100644 --- a/source/extensions/extensions_build_config.bzl +++ b/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 diff --git a/source/extensions/stat_sinks/hystrix/BUILD b/source/extensions/stat_sinks/hystrix/BUILD new file mode 100644 index 0000000000000..ccf336a5e7f1d --- /dev/null +++ b/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", + ], +) diff --git a/source/extensions/stat_sinks/hystrix/config.cc b/source/extensions/stat_sinks/hystrix/config.cc new file mode 100644 index 0000000000000..8c547c4417cba --- /dev/null +++ b/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(config); + if (hystrix_sink.num_of_buckets() == 0) { // if not set + return std::make_unique(server); + } + return std::make_unique(server, hystrix_sink.num_of_buckets()); +} + +ProtobufTypes::MessagePtr HystrixSinkFactory::createEmptyConfigProto() { + return std::unique_ptr( + 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 + register_; + +} // namespace Hystrix +} // namespace StatSinks +} // namespace Extensions +} // namespace Envoy diff --git a/source/extensions/stat_sinks/hystrix/config.h b/source/extensions/stat_sinks/hystrix/config.h new file mode 100644 index 0000000000000..2f3f7c37f8783 --- /dev/null +++ b/source/extensions/stat_sinks/hystrix/config.h @@ -0,0 +1,29 @@ +#pragma once + +#include + +#include "envoy/server/instance.h" + +#include "server/configuration_impl.h" + +namespace Envoy { +namespace Extensions { +namespace StatSinks { +namespace Hystrix { + +class HystrixSinkFactory : Logger::Loggable, + 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 diff --git a/source/extensions/stat_sinks/hystrix/hystrix.cc b/source/extensions/stat_sinks/hystrix/hystrix.cc new file mode 100644 index 0000000000000..8a7f441f9d85e --- /dev/null +++ b/source/extensions/stat_sinks/hystrix/hystrix.cc @@ -0,0 +1,371 @@ +#include "extensions/stat_sinks/hystrix/hystrix.h" + +#include +#include +#include +#include + +#include "common/buffer/buffer_impl.h" +#include "common/common/logger.h" + +#include "server/http/admin.h" + +#include "absl/strings/str_cat.h" + +namespace Envoy { +namespace Extensions { +namespace StatSinks { + +const uint64_t HystrixStatCache::DEFAULT_NUM_OF_BUCKETS; + +// Add new value to rolling window, in place of oldest one. +void HystrixStatCache::pushNewValue(const std::string& key, uint64_t value) { + // Create vector if do not exist. + // TODO trabetti: why resize + value param didn't work without the if? + if (rolling_stats_map_.find(key) == rolling_stats_map_.end()) { + rolling_stats_map_[key].resize(window_size_, value); + } else { + rolling_stats_map_[key][current_index_] = value; + } +} + +uint64_t HystrixStatCache::getRollingValue(absl::string_view cluster_name, absl::string_view stat) { + std::string key = absl::StrCat("cluster.", cluster_name, ".", stat); + if (rolling_stats_map_.find(key) != rolling_stats_map_.end()) { + // If the counter was reset, the result is negative + // better return 0, will be back to normal once one rolling window passes. + if (rolling_stats_map_[key][current_index_] < + rolling_stats_map_[key][(current_index_ + 1) % window_size_]) { + return 0; + } else { + return rolling_stats_map_[key][current_index_] - + rolling_stats_map_[key][(current_index_ + 1) % window_size_]; + } + } else { + return 0; + } +} + +void HystrixStatCache::CreateCounterNameLookupForCluster(const std::string& cluster_name) { + // Building lookup name map for all specific cluster values. + // Every call to the updateRollingWindowMap function should get the appropriate name from the map. + std::string cluster_name_with_prefix = absl::StrCat("cluster.", cluster_name, "."); + counter_name_lookup[cluster_name]["upstream_rq_5xx"] = + absl::StrCat(cluster_name_with_prefix, "upstream_rq_5xx"); + counter_name_lookup[cluster_name]["retry.upstream_rq_5xx"] = + absl::StrCat(cluster_name_with_prefix, "retry.upstream_rq_5xx"); + counter_name_lookup[cluster_name]["upstream_rq_4xx"] = + absl::StrCat(cluster_name_with_prefix, "upstream_rq_4xx"); + counter_name_lookup[cluster_name]["retry.upstream_rq_4xx"] = + absl::StrCat(cluster_name_with_prefix, "retry.upstream_rq_4xx"); + counter_name_lookup[cluster_name]["errors"] = absl::StrCat(cluster_name_with_prefix, "errors"); + counter_name_lookup[cluster_name]["upstream_rq_2xx"] = + absl::StrCat(cluster_name_with_prefix, "upstream_rq_2xx"); + counter_name_lookup[cluster_name]["success"] = absl::StrCat(cluster_name_with_prefix, "success"); + counter_name_lookup[cluster_name]["rejected"] = + absl::StrCat(cluster_name_with_prefix, "rejected"); + counter_name_lookup[cluster_name]["timeouts"] = + absl::StrCat(cluster_name_with_prefix, "timeouts"); + counter_name_lookup[cluster_name]["total"] = absl::StrCat(cluster_name_with_prefix, "total"); +} + +void HystrixStatCache::updateRollingWindowMap(Upstream::ClusterInfoConstSharedPtr cluster_info, + Stats::Store& stats) { + std::string cluster_name = cluster_info->name(); + Upstream::ClusterStats& cluster_stats = cluster_info->stats(); + + if (counter_name_lookup.find(cluster_name) == counter_name_lookup.end()) { + CreateCounterNameLookupForCluster(cluster_name); + } + + // Combining timeouts+retries - retries are counted as separate requests + // (alternative: each request including the retries counted as 1). + uint64_t timeouts = cluster_stats.upstream_rq_timeout_.value() + + cluster_stats.upstream_rq_per_try_timeout_.value(); + + pushNewValue(counter_name_lookup[cluster_name]["timeouts"], timeouts); + + // Combining errors+retry errors - retries are counted as separate requests + // (alternative: each request including the retries counted as 1) + // since timeouts are 504 (or 408), deduce them from here ("-" sign). + // Timeout retries were not counted here anyway. + uint64_t errors = + stats.counter(counter_name_lookup[cluster_name]["upstream_rq_5xx"]).value() + + stats.counter(counter_name_lookup[cluster_name]["retry.upstream_rq_5xx"]).value() + + stats.counter(counter_name_lookup[cluster_name]["upstream_rq_4xx"]).value() + + stats.counter(counter_name_lookup[cluster_name]["retry.upstream_rq_4xx"]).value() - + cluster_stats.upstream_rq_timeout_.value(); + + pushNewValue(counter_name_lookup[cluster_name]["errors"], errors); + + uint64_t success = stats.counter(counter_name_lookup[cluster_name]["upstream_rq_2xx"]).value(); + pushNewValue(counter_name_lookup[cluster_name]["success"], success); + + uint64_t rejected = cluster_stats.upstream_rq_pending_overflow_.value(); + pushNewValue(counter_name_lookup[cluster_name]["rejected"], rejected); + + // should not take from upstream_rq_total since it is updated before its components, + // leading to wrong results such as error percentage higher than 100% + uint64_t total = errors + timeouts + success + rejected; + pushNewValue(counter_name_lookup[cluster_name]["total"], total); + + ENVOY_LOG(trace, "{}", printRollingWindow()); +} + +void HystrixStatCache::resetRollingWindow() { rolling_stats_map_.clear(); } + +void HystrixStatCache::addStringToStream(absl::string_view key, absl::string_view value, + std::stringstream& info) { + std::string quoted_value = absl::StrCat("\"", value, "\""); + addInfoToStream(key, quoted_value, info); +} + +void HystrixStatCache::addIntToStream(absl::string_view key, uint64_t value, + std::stringstream& info) { + addInfoToStream(key, std::to_string(value), info); +} + +void HystrixStatCache::addInfoToStream(absl::string_view key, absl::string_view value, + std::stringstream& info) { + if (!info.str().empty()) { + info << ", "; + } + std::string added_info = absl::StrCat("\"", key, "\": ", value); + info << added_info; +} + +void HystrixStatCache::addHystrixCommand(std::stringstream& ss, absl::string_view cluster_name, + uint64_t max_concurrent_requests, uint64_t reporting_hosts, + uint64_t rolling_window) { + std::stringstream cluster_info; + std::time_t currentTime = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()); + addStringToStream("type", "HystrixCommand", cluster_info); + addStringToStream("name", cluster_name, cluster_info); + addStringToStream("group", "NA", cluster_info); + addIntToStream("currentTime", static_cast(currentTime), cluster_info); + addInfoToStream("isCircuitBreakerOpen", "false", cluster_info); + + uint64_t errors = getRollingValue(cluster_name, "errors"); + uint64_t timeouts = getRollingValue(cluster_name, "timeouts"); + uint64_t rejected = getRollingValue(cluster_name, "rejected"); + uint64_t total = getRollingValue(cluster_name, "total"); + + uint64_t error_rate = + total == 0 + ? 0 + : (static_cast(errors + timeouts + rejected) / static_cast(total)) * 100; + + addIntToStream("errorPercentage", error_rate, cluster_info); + addIntToStream("errorCount", errors, cluster_info); + addIntToStream("requestCount", total, cluster_info); + addIntToStream("rollingCountCollapsedRequests", 0, cluster_info); + addIntToStream("rollingCountExceptionsThrown", 0, cluster_info); + addIntToStream("rollingCountFailure", errors, cluster_info); + addIntToStream("rollingCountFallbackFailure", 0, cluster_info); + addIntToStream("rollingCountFallbackRejection", 0, cluster_info); + addIntToStream("rollingCountFallbackSuccess", 0, cluster_info); + addIntToStream("rollingCountResponsesFromCache", 0, cluster_info); + + // Envoy's "circuit breaker" has similar meaning to hystrix's isolation + // so we count upstream_rq_pending_overflow and present it as rejected + addIntToStream("rollingCountSemaphoreRejected", rejected, cluster_info); + + // Hystrix's short circuit is not similar to Envoy's since it is triggered by 503 responses + // there is no parallel counter in Envoy since as a result of errors (outlier detection) + // requests are not rejected, but rather the node is removed from load balancer healthy pool. + addIntToStream("rollingCountShortCircuited", 0, cluster_info); + addIntToStream("rollingCountSuccess", getRollingValue(cluster_name, "success"), cluster_info); + addIntToStream("rollingCountThreadPoolRejected", 0, cluster_info); + addIntToStream("rollingCountTimeout", timeouts, cluster_info); + addIntToStream("rollingCountBadRequests", 0, cluster_info); + addIntToStream("currentConcurrentExecutionCount", 0, cluster_info); + addIntToStream("latencyExecute_mean", 0, cluster_info); + + // 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}", + cluster_info); + addIntToStream("propertyValue_circuitBreakerRequestVolumeThreshold", 0, cluster_info); + addIntToStream("propertyValue_circuitBreakerSleepWindowInMilliseconds", 0, cluster_info); + addIntToStream("propertyValue_circuitBreakerErrorThresholdPercentage", 0, cluster_info); + addInfoToStream("propertyValue_circuitBreakerForceOpen", "false", cluster_info); + addInfoToStream("propertyValue_circuitBreakerForceClosed", "true", cluster_info); + addStringToStream("propertyValue_executionIsolationStrategy", "SEMAPHORE", cluster_info); + addIntToStream("propertyValue_executionIsolationThreadTimeoutInMilliseconds", 0, cluster_info); + addInfoToStream("propertyValue_executionIsolationThreadInterruptOnTimeout", "false", + cluster_info); + addIntToStream("propertyValue_executionIsolationSemaphoreMaxConcurrentRequests", + max_concurrent_requests, cluster_info); + addIntToStream("propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests", 0, cluster_info); + addInfoToStream("propertyValue_requestCacheEnabled", "false", cluster_info); + addInfoToStream("propertyValue_requestLogEnabled", "true", cluster_info); + addIntToStream("reportingHosts", reporting_hosts, cluster_info); + addIntToStream("propertyValue_metricsRollingStatisticalWindowInMilliseconds", rolling_window, + cluster_info); + + ss << "data: {" << cluster_info.str() << "}" << std::endl << std::endl; +} + +void HystrixStatCache::addHystrixThreadPool(std::stringstream& ss, absl::string_view cluster_name, + uint64_t queue_size, uint64_t reporting_hosts, + uint64_t rolling_window) { + std::stringstream cluster_info; + + addIntToStream("currentPoolSize", 0, cluster_info); + addIntToStream("rollingMaxActiveThreads", 0, cluster_info); + addIntToStream("currentActiveCount", 0, cluster_info); + addIntToStream("currentCompletedTaskCount", 0, cluster_info); + addIntToStream("propertyValue_queueSizeRejectionThreshold", queue_size, cluster_info); + addStringToStream("type", "HystrixThreadPool", cluster_info); + addIntToStream("reportingHosts", reporting_hosts, cluster_info); + addIntToStream("propertyValue_metricsRollingStatisticalWindowInMilliseconds", rolling_window, + cluster_info); + addStringToStream("name", cluster_name, cluster_info); + addIntToStream("currentLargestPoolSize", 0, cluster_info); + addIntToStream("currentCorePoolSize", 0, cluster_info); + addIntToStream("currentQueueSize", 0, cluster_info); + addIntToStream("currentTaskCount", 0, cluster_info); + addIntToStream("rollingCountThreadsExecuted", 0, cluster_info); + addIntToStream("currentMaximumPoolSize", 0, cluster_info); + + ss << "data: {" << cluster_info.str() << "}" << std::endl << std::endl; +} + +void HystrixStatCache::getClusterStats(std::stringstream& ss, absl::string_view cluster_name, + uint64_t max_concurrent_requests, uint64_t reporting_hosts, + uint64_t rolling_window) { + addHystrixCommand(ss, cluster_name, max_concurrent_requests, reporting_hosts, rolling_window); + addHystrixThreadPool(ss, cluster_name, max_concurrent_requests, reporting_hosts, rolling_window); +} + +const std::string HystrixStatCache::printRollingWindow() const { + std::stringstream out_str; + + for (auto stats_map_itr = rolling_stats_map_.begin(); stats_map_itr != rolling_stats_map_.end(); + ++stats_map_itr) { + out_str << stats_map_itr->first << " | "; + RollingWindow rolling_window = stats_map_itr->second; + for (auto specific_stat_vec_itr = rolling_window.begin(); + specific_stat_vec_itr != rolling_window.end(); ++specific_stat_vec_itr) { + out_str << *specific_stat_vec_itr << " | "; + } + out_str << std::endl; + } + return out_str.str(); +} + +namespace Hystrix { +HystrixSink::HystrixSink(Server::Instance& server, const uint64_t num_of_buckets) + : stats_(new HystrixStatCache(num_of_buckets)), server_(server) { + init(); +} + +HystrixSink::HystrixSink(Server::Instance& server) + : stats_(new HystrixStatCache()), server_(server) { + init(); +} + +void HystrixSink::init() { + Server::Admin& admin = server_.admin(); + ENVOY_LOG(debug, + "adding hystrix_event_stream endpoint to enable connection to hystrix dashboard"); + admin.addHandler("/hystrix_event_stream", "send hystrix event stream", + MAKE_ADMIN_HANDLER(handlerHystrixEventStream), false, false); +} + +Http::Code HystrixSink::handlerHystrixEventStream(absl::string_view, + Http::HeaderMap& response_headers, + Buffer::Instance&, + Server::AdminFilter& admin_filter) { + + response_headers.insertContentType().value().setReference( + Http::Headers::get().ContentTypeValues.TextEventStream); + response_headers.insertCacheControl().value().setReference( + Http::Headers::get().CacheControlValues.NoCache); + response_headers.insertConnection().value().setReference( + Http::Headers::get().ConnectionValues.Close); + response_headers.insertAccessControlAllowHeaders().value().setReference( + Http::Headers::get().AccessControlAllowHeadersValue.AccessControlAllowHeadersHystrix); + response_headers.insertAccessControlAllowOrigin().value().setReference( + Http::Headers::get().AccessControlAllowOriginValue.All); + response_headers.insertNoChunks().value().setReference("0"); + + Http::StreamDecoderFilterCallbacks* stream_decoder_filter_callbacks = + admin_filter.getDecoderFilterCallbacks(); + + registerConnection(stream_decoder_filter_callbacks); + + admin_filter.setEndStreamOnComplete(false); // set streaming + + // Separated out just so it's easier to understand + auto on_destroy_callback = [this, stream_decoder_filter_callbacks]() { + // Unregister the callbacks from the sink so data is no longer encoded through them. + unregisterConnection(stream_decoder_filter_callbacks); + }; + + // Add the callback to the admin_filter list of callbacks + admin_filter.addOnDestroyCallback(std::move(on_destroy_callback)); + + ENVOY_LOG(debug, "start sending data to hystrix dashboard on port {}", + stream_decoder_filter_callbacks->connection()->localAddress()->asString()); + return Http::Code::OK; +} + +void HystrixSink::beginFlush() { current_stat_values_.clear(); } + +void HystrixSink::endFlush() { + if (callbacks_list_.empty()) + return; + stats_->incCounter(); + for (auto& cluster : server_.clusterManager().clusters()) { + stats_->updateRollingWindowMap(cluster.second.get().info(), server_.stats()); + } + std::stringstream ss; + for (auto& cluster : server_.clusterManager().clusters()) { + stats_->getClusterStats( + ss, cluster.second.get().info()->name(), + cluster.second.get() + .info() + ->resourceManager(Upstream::ResourcePriority::Default) + .pendingRequests() + .max(), + server_.stats() + .gauge("cluster." + cluster.second.get().info()->name() + ".membership_total") + .value(), + server_.statsFlushInterval().count()); + } + Buffer::OwnedImpl data; + for (auto callbacks : callbacks_list_) { + data.add(ss.str()); + callbacks->encodeData(data, false); + } + + // send keep alive ping + // TODO (@trabetti) : is it ok to send together with data? + Buffer::OwnedImpl ping_data; + for (auto callbacks : callbacks_list_) { + ping_data.add(":\n\n"); + callbacks->encodeData(ping_data, false); + } +} + +void HystrixSink::registerConnection(Http::StreamDecoderFilterCallbacks* callbacks_to_register) { + callbacks_list_.emplace_back(callbacks_to_register); +} + +void HystrixSink::unregisterConnection(Http::StreamDecoderFilterCallbacks* callbacks_to_remove) { + for (auto it = callbacks_list_.begin(); it != callbacks_list_.end();) { + if ((*it)->streamId() == callbacks_to_remove->streamId()) { + it = callbacks_list_.erase(it); + break; + } else { + ++it; + } + } +} + +} // namespace Hystrix +} // namespace StatSinks +} // namespace Extensions +} // namespace Envoy diff --git a/source/extensions/stat_sinks/hystrix/hystrix.h b/source/extensions/stat_sinks/hystrix/hystrix.h new file mode 100644 index 0000000000000..3b20607d22a64 --- /dev/null +++ b/source/extensions/stat_sinks/hystrix/hystrix.h @@ -0,0 +1,144 @@ +#include +#include +#include + +#include "envoy/server/admin.h" +#include "envoy/server/instance.h" +#include "envoy/stats/stats.h" + +namespace Envoy { +namespace Extensions { +namespace StatSinks { + +typedef std::vector RollingWindow; +typedef std::map RollingStatsMap; + +class HystrixStatCache : public Logger::Loggable { +public: + HystrixStatCache() + : current_index_(DEFAULT_NUM_OF_BUCKETS), window_size_(DEFAULT_NUM_OF_BUCKETS + 1){}; + + HystrixStatCache(uint64_t num_of_buckets) + : current_index_(num_of_buckets), window_size_(num_of_buckets + 1){}; + + /** + * Add new value to top of rolling window, pushing out the oldest value. + */ + void pushNewValue(const std::string& key, uint64_t value); + + /** + * Increment pointer of next value to add to rolling window. + */ + void incCounter() { current_index_ = (current_index_ + 1) % window_size_; } + + /** + * Generate the streams to be sent to hystrix dashboard. + */ + void getClusterStats(std::stringstream& ss, absl::string_view cluster_name, + uint64_t max_concurrent_requests, uint64_t reporting_hosts, + uint64_t rolling_window); + + /** + * Calculate values needed to create the stream and write into the map. + */ + void updateRollingWindowMap(Upstream::ClusterInfoConstSharedPtr cluster_info, + Stats::Store& stats); + /** + * Clear map. + */ + void resetRollingWindow(); + + /** + * Return string represnting current state of the map. for DEBUG. + */ + const std::string printRollingWindow() const; + + /** + * Get the statistic's value change over the rolling window time frame. + */ + uint64_t getRollingValue(absl::string_view cluster_name, absl::string_view stat); + +private: + /** + * Format the given key and absl::string_view value to "key"="value", and adding to the + * stringstream. + */ + void addStringToStream(absl::string_view key, absl::string_view value, std::stringstream& info); + + /** + * Format the given key and uint64_t value to "key"=, and adding to the + * stringstream. + */ + void addIntToStream(absl::string_view key, uint64_t value, std::stringstream& info); + + /** + * Format the given key and value to "key"=value, and adding to the stringstream. + */ + void addInfoToStream(absl::string_view key, absl::string_view value, std::stringstream& info); + + /** + * Generate HystrixCommand event stream. + */ + void addHystrixCommand(std::stringstream& ss, absl::string_view cluster_name, + uint64_t max_concurrent_requests, uint64_t reporting_hosts, + uint64_t rolling_window); + + /** + * Generate HystrixThreadPool event stream. + */ + void addHystrixThreadPool(std::stringstream& ss, absl::string_view cluster_name, + uint64_t queue_size, uint64_t reporting_hosts, uint64_t rolling_window); + + /** + * Building lookup name map for all specific cluster values. + */ + void CreateCounterNameLookupForCluster(const std::string& cluster_name); + + RollingStatsMap rolling_stats_map_; + uint64_t current_index_; + const uint64_t window_size_; + // TODO(trabetti): do we want this to be configurable through the HystrixSink in config file? + static const uint64_t DEFAULT_NUM_OF_BUCKETS = 10; + std::map> counter_name_lookup; +}; + +typedef std::unique_ptr HystrixStatCachePtr; + +namespace Hystrix { + +class HystrixSink : public Stats::Sink, public Logger::Loggable { +public: + HystrixSink(Server::Instance& server, uint64_t num_of_buckets); + HystrixSink(Server::Instance& server); + Http::Code handlerHystrixEventStream(absl::string_view, Http::HeaderMap& response_headers, + Buffer::Instance&, Server::AdminFilter& admin_filter); + void init(); + void beginFlush() override; + void flushCounter(const Stats::Counter&, uint64_t) override{}; + void flushGauge(const Stats::Gauge&, uint64_t) override{}; + void endFlush() override; + void onHistogramComplete(const Stats::Histogram&, uint64_t) override{}; + + /** + * register a new connection + */ + void registerConnection(Http::StreamDecoderFilterCallbacks* callbacks_to_register); + /** + * remove registered connection + */ + void unregisterConnection(Http::StreamDecoderFilterCallbacks* callbacks_to_remove); + HystrixStatCache& getStats() { return *stats_; } + +private: + HystrixStatCachePtr stats_; + std::vector callbacks_list_{}; + Server::Instance& server_; + std::map current_stat_values_; +}; + +typedef std::unique_ptr HystrixSinkPtr; + +} // namespace Hystrix +} // namespace StatSinks +} // namespace Extensions +} // namespace Envoy diff --git a/source/extensions/stat_sinks/well_known_names.h b/source/extensions/stat_sinks/well_known_names.h index f524ffa4cd181..5f98e3c096246 100644 --- a/source/extensions/stat_sinks/well_known_names.h +++ b/source/extensions/stat_sinks/well_known_names.h @@ -18,6 +18,8 @@ class StatsSinkNameValues { const std::string DOG_STATSD = "envoy.dog_statsd"; // MetricsService sink const std::string METRICS_SERVICE = "envoy.metrics_service"; + // Hystrix sink + const std::string HYSTRIX = "envoy.hystrix"; }; typedef ConstSingleton StatsSinkNames; diff --git a/source/server/BUILD b/source/server/BUILD index 25fbcb8692176..f449ccc2cde5b 100644 --- a/source/server/BUILD +++ b/source/server/BUILD @@ -272,6 +272,8 @@ envoy_cc_library( "//source/common/singleton:manager_impl_lib", "//source/common/stats:thread_local_store_lib", "//source/common/upstream:cluster_manager_lib", + "//source/extensions/stat_sinks/common/statsd:statsd_lib", + "//source/extensions/stat_sinks/hystrix:hystrix_lib", "//source/server/http:admin_lib", "@envoy_api//envoy/config/bootstrap/v2:bootstrap_cc", ], diff --git a/source/server/config_validation/server.h b/source/server/config_validation/server.h index 4085c9b45e025..fc674992d8a16 100644 --- a/source/server/config_validation/server.h +++ b/source/server/config_validation/server.h @@ -89,6 +89,8 @@ class ValidationInstance : Logger::Loggable, ThreadLocal::Instance& threadLocal() override { return thread_local_; } const LocalInfo::LocalInfo& localInfo() override { return *local_info_; } + std::chrono::milliseconds statsFlushInterval() override { return config_->statsFlushInterval(); } + // Server::ListenerComponentFactory std::vector createNetworkFilterFactoryList( const Protobuf::RepeatedPtrField& filters, diff --git a/source/server/http/admin.cc b/source/server/http/admin.cc index 6cb19190dc222..6ff4aaf104ca1 100644 --- a/source/server/http/admin.cc +++ b/source/server/http/admin.cc @@ -150,6 +150,16 @@ Http::FilterTrailersStatus AdminFilter::decodeTrailers(Http::HeaderMap&) { return Http::FilterTrailersStatus::StopIteration; } +void AdminFilter::onDestroy() { + for (const auto& callback : on_destroy_callbacks_) { + callback(); + } +} + +void AdminFilter::addOnDestroyCallback(std::function cb) { + on_destroy_callbacks_.push_back(std::move(cb)); +} + bool AdminImpl::changeLogLevel(const Http::Utility::QueryParams& params) { if (params.size() != 1) { return false; @@ -222,7 +232,7 @@ void AdminImpl::addCircuitSettings(const std::string& cluster_name, const std::s } Http::Code AdminImpl::handlerClusters(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { response.add(fmt::format("version_info::{}\n", server_.clusterManager().versionInfo())); for (auto& cluster : server_.clusterManager().clusters()) { @@ -280,7 +290,7 @@ Http::Code AdminImpl::handlerClusters(absl::string_view, Http::HeaderMap&, // TODO(jsedgwick) Use query params to list available dumps, selectively dump, etc Http::Code AdminImpl::handlerConfigDump(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) const { + Buffer::Instance& response, AdminFilter&) const { envoy::admin::v2::ConfigDump dump; auto& config_dump_map = *(dump.mutable_configs()); for (const auto& key_callback_pair : config_tracker_.getCallbacksMap()) { @@ -296,7 +306,7 @@ Http::Code AdminImpl::handlerConfigDump(absl::string_view, Http::HeaderMap&, } Http::Code AdminImpl::handlerCpuProfiler(absl::string_view url, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { Http::Utility::QueryParams query_params = Http::Utility::parseQueryString(url); if (query_params.size() != 1 || query_params.begin()->first != "enable" || (query_params.begin()->second != "y" && query_params.begin()->second != "n")) { @@ -320,27 +330,27 @@ Http::Code AdminImpl::handlerCpuProfiler(absl::string_view url, Http::HeaderMap& } Http::Code AdminImpl::handlerHealthcheckFail(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { server_.failHealthcheck(true); response.add("OK\n"); return Http::Code::OK; } Http::Code AdminImpl::handlerHealthcheckOk(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { server_.failHealthcheck(false); response.add("OK\n"); return Http::Code::OK; } Http::Code AdminImpl::handlerHotRestartVersion(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { response.add(server_.hotRestart().version()); return Http::Code::OK; } Http::Code AdminImpl::handlerLogging(absl::string_view url, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { Http::Utility::QueryParams query_params = Http::Utility::parseQueryString(url); Http::Code rc = Http::Code::OK; @@ -366,7 +376,7 @@ Http::Code AdminImpl::handlerLogging(absl::string_view url, Http::HeaderMap&, } Http::Code AdminImpl::handlerResetCounters(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { for (const Stats::CounterSharedPtr& counter : server_.stats().counters()) { counter->reset(); } @@ -376,7 +386,7 @@ Http::Code AdminImpl::handlerResetCounters(absl::string_view, Http::HeaderMap&, } Http::Code AdminImpl::handlerServerInfo(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { time_t current_time = time(nullptr); response.add(fmt::format("envoy {} {} {} {} {}\n", VersionInfo::version(), server_.healthCheckFailed() ? "draining" : "live", @@ -387,7 +397,7 @@ Http::Code AdminImpl::handlerServerInfo(absl::string_view, Http::HeaderMap&, } Http::Code AdminImpl::handlerStats(absl::string_view url, Http::HeaderMap& response_headers, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter& admin_filter) { // We currently don't support timers locally (only via statsd) so just group all the counters // and gauges together, alpha sort them, and spit them out. Http::Code rc = Http::Code::OK; @@ -414,7 +424,7 @@ Http::Code AdminImpl::handlerStats(absl::string_view url, Http::HeaderMap& respo Http::Headers::get().ContentTypeValues.Json); response.add(AdminImpl::statsAsJson(all_stats)); } else if (format_key == "format" && format_value == "prometheus") { - return handlerPrometheusStats(url, response_headers, response); + return handlerPrometheusStats(url, response_headers, response, admin_filter); } else { response.add("usage: /stats?format=json \n"); response.add("\n"); @@ -425,7 +435,7 @@ Http::Code AdminImpl::handlerStats(absl::string_view url, Http::HeaderMap& respo } Http::Code AdminImpl::handlerPrometheusStats(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { PrometheusStatsFormatter::statsAsPrometheus(server_.stats().counters(), server_.stats().gauges(), response); return Http::Code::OK; @@ -504,14 +514,14 @@ std::string AdminImpl::statsAsJson(const std::map& all_st } Http::Code AdminImpl::handlerQuitQuitQuit(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { server_.shutdown(); response.add("OK\n"); return Http::Code::OK; } Http::Code AdminImpl::handlerListenerInfo(absl::string_view, Http::HeaderMap& response_headers, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { response_headers.insertContentType().value().setReference( Http::Headers::get().ContentTypeValues.Json); std::list listeners; @@ -522,8 +532,8 @@ Http::Code AdminImpl::handlerListenerInfo(absl::string_view, Http::HeaderMap& re return Http::Code::OK; } -Http::Code AdminImpl::handlerCerts(absl::string_view, Http::HeaderMap&, - Buffer::Instance& response) { +Http::Code AdminImpl::handlerCerts(absl::string_view, Http::HeaderMap&, Buffer::Instance& response, + AdminFilter&) { // This set is used to track distinct certificates. We may have multiple listeners, upstreams, etc // using the same cert. std::unordered_set context_info_set; @@ -542,7 +552,7 @@ Http::Code AdminImpl::handlerCerts(absl::string_view, Http::HeaderMap&, } Http::Code AdminImpl::handlerRuntime(absl::string_view url, Http::HeaderMap& response_headers, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { const Http::Utility::QueryParams params = Http::Utility::parseQueryString(url); response_headers.insertContentType().value().setReference( Http::Headers::get().ContentTypeValues.Json); @@ -630,7 +640,7 @@ std::string AdminImpl::runtimeAsJson( } Http::Code AdminImpl::handlerRuntimeModify(absl::string_view url, Http::HeaderMap&, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { const Http::Utility::QueryParams params = Http::Utility::parseQueryString(url); if (params.empty()) { response.add("usage: /runtime_modify?key1=value1&key2=value2&keyN=valueN\n"); @@ -653,7 +663,8 @@ void AdminFilter::onComplete() { Buffer::OwnedImpl response; Http::HeaderMapPtr header_map{new Http::HeaderMapImpl}; RELEASE_ASSERT(request_headers_); - Http::Code code = parent_.runCallback(path, *request_headers_, *header_map, response); + Http::Code code = parent_.runCallback(path, *header_map, response, *this); + header_map->insertStatus().value(std::to_string(enumToInt(code))); const auto& headers = Http::Headers::get(); if (header_map->ContentType() == nullptr) { @@ -668,10 +679,11 @@ void AdminFilter::onComplete() { // Under no circumstance should browsers sniff content-type. header_map->addReference(headers.XContentTypeOptions, headers.XContentTypeOptionValues.Nosniff); - callbacks_->encodeHeaders(std::move(header_map), response.length() == 0); + callbacks_->encodeHeaders(std::move(header_map), + end_stream_on_complete_ && response.length() == 0); if (response.length() > 0) { - callbacks_->encodeData(response, true); + callbacks_->encodeData(response, end_stream_on_complete_); } } @@ -702,7 +714,7 @@ AdminImpl::AdminImpl(const std::string& access_log_path, const std::string& prof MAKE_ADMIN_HANDLER(handlerHealthcheckOk), false, true}, {"/help", "print out list of admin commands", MAKE_ADMIN_HANDLER(handlerHelp), false, false}, - {"/hot_restart_version", "print the hot restart compatability version", + {"/hot_restart_version", "print the hot restart compatibility version", MAKE_ADMIN_HANDLER(handlerHotRestartVersion), false, false}, {"/logging", "query/change logging levels", MAKE_ADMIN_HANDLER(handlerLogging), false, true}, @@ -719,7 +731,9 @@ AdminImpl::AdminImpl(const std::string& access_log_path, const std::string& prof false}, {"/runtime", "print runtime values", MAKE_ADMIN_HANDLER(handlerRuntime), false, false}, {"/runtime_modify", "modify runtime values", MAKE_ADMIN_HANDLER(handlerRuntimeModify), - false, true}}, + false, true}, + }, + // TODO(jsedgwick) add /runtime_reset endpoint that removes all admin-set values listener_(*this, std::move(listener_scope)) { @@ -759,8 +773,8 @@ void AdminImpl::createFilterChain(Http::FilterChainFactoryCallbacks& callbacks) } Http::Code AdminImpl::runCallback(absl::string_view path_and_query, - const Http::HeaderMap& request_headers, - Http::HeaderMap& response_headers, Buffer::Instance& response) { + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter& admin_filter) { Http::Code code = Http::Code::OK; bool found_handler = false; @@ -772,13 +786,14 @@ Http::Code AdminImpl::runCallback(absl::string_view path_and_query, for (const UrlHandler& handler : handlers_) { if (path_and_query.compare(0, query_index, handler.prefix_) == 0) { if (handler.mutates_server_state_) { - const absl::string_view method = request_headers.Method()->value().getStringView(); + const absl::string_view method = + admin_filter.getRequestHeaders()->Method()->value().getStringView(); if (method != Http::Headers::get().MethodValues.Post) { ENVOY_LOG(warn, "admin path \"{}\" mutates state, method={} rather than POST", handler.prefix_, method); } } - code = handler.handler_(path_and_query, response_headers, response); + code = handler.handler_(path_and_query, response_headers, response, admin_filter); found_handler = true; break; } @@ -788,7 +803,7 @@ Http::Code AdminImpl::runCallback(absl::string_view path_and_query, // Extra space is emitted below to have "invalid path." be a separate sentence in the // 404 output from "admin commands are:" in handlerHelp. response.add("invalid path. "); - handlerHelp(path_and_query, response_headers, response); + handlerHelp(path_and_query, response_headers, response, admin_filter); code = Http::Code::NotFound; } @@ -806,7 +821,8 @@ std::vector AdminImpl::sortedHandlers() const { return sorted_handlers; } -Http::Code AdminImpl::handlerHelp(absl::string_view, Http::HeaderMap&, Buffer::Instance& response) { +Http::Code AdminImpl::handlerHelp(absl::string_view, Http::HeaderMap&, Buffer::Instance& response, + AdminFilter&) { response.add("admin commands are:\n"); // Prefix order is used during searching, but for printing do them in alpha order. @@ -817,7 +833,7 @@ Http::Code AdminImpl::handlerHelp(absl::string_view, Http::HeaderMap&, Buffer::I } Http::Code AdminImpl::handlerAdminHome(absl::string_view, Http::HeaderMap& response_headers, - Buffer::Instance& response) { + Buffer::Instance& response, AdminFilter&) { response_headers.insertContentType().value().setReference( Http::Headers::get().ContentTypeValues.Html); diff --git a/source/server/http/admin.h b/source/server/http/admin.h index 18409fc15c256..80fa895b1ba1d 100644 --- a/source/server/http/admin.h +++ b/source/server/http/admin.h @@ -31,6 +31,7 @@ namespace Envoy { namespace Server { +class AdminFilter; /** * Implementation of Server::Admin. */ @@ -44,8 +45,8 @@ class AdminImpl : public Admin, const std::string& address_out_path, Network::Address::InstanceConstSharedPtr address, Server::Instance& server, Stats::ScopePtr&& listener_scope); - Http::Code runCallback(absl::string_view path_and_query, const Http::HeaderMap& request_headers, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::Code runCallback(absl::string_view path_and_query, Http::HeaderMap& response_headers, + Buffer::Instance& response, AdminFilter& admin_filter); const Network::Socket& socket() override { return *socket_; } Network::Socket& mutable_socket() { return *socket_; } Network::ListenerConfig& listener() { return listener_; } @@ -144,43 +145,50 @@ class AdminImpl : public Admin, * URL handlers. */ Http::Code handlerAdminHome(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerCerts(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerClusters(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerConfigDump(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response) const; + Buffer::Instance& response, AdminFilter&) const; Http::Code handlerCpuProfiler(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerHealthcheckFail(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerHealthcheckOk(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerHelp(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerHotRestartVersion(absl::string_view path_and_query, - Http::HeaderMap& response_headers, - Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerListenerInfo(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerLogging(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); - Http::Code handlerMain(const std::string& path, Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); + Http::Code handlerMain(const std::string& path, Buffer::Instance& response, AdminFilter&); Http::Code handlerQuitQuitQuit(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerResetCounters(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerServerInfo(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerStats(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerPrometheusStats(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); Http::Code handlerRuntime(absl::string_view path_and_query, Http::HeaderMap& response_headers, - Buffer::Instance& response); + Buffer::Instance& response, AdminFilter&); Http::Code handlerRuntimeModify(absl::string_view path_and_query, - Http::HeaderMap& response_headers, Buffer::Instance& response); + Http::HeaderMap& response_headers, Buffer::Instance& response, + AdminFilter&); class AdminListener : public Network::ListenerConfig { public: @@ -233,8 +241,10 @@ class AdminFilter : public Http::StreamDecoderFilter, Logger::Loggable cb); + Http::StreamDecoderFilterCallbacks* getDecoderFilterCallbacks() { return callbacks_; } + const Http::HeaderMap* getRequestHeaders() { return request_headers_; } // Http::StreamDecoderFilter Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override; Http::FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream) override; @@ -242,6 +252,7 @@ class AdminFilter : public Http::StreamDecoderFilter, Logger::Loggable> on_destroy_callbacks_; + bool end_stream_on_complete_ = true; }; /** diff --git a/source/server/server.cc b/source/server/server.cc index 74ea43d58989d..35e809a1adc36 100644 --- a/source/server/server.cc +++ b/source/server/server.cc @@ -37,6 +37,8 @@ #include "server/guarddog_impl.h" #include "server/test_hooks.h" +#include "extensions/stat_sinks/hystrix/hystrix.h" + namespace Envoy { namespace Server { diff --git a/source/server/server.h b/source/server/server.h index 645f0a0067463..d077d776ff829 100644 --- a/source/server/server.h +++ b/source/server/server.h @@ -168,6 +168,8 @@ class InstanceImpl : Logger::Loggable, public Instance { ThreadLocal::Instance& threadLocal() override { return thread_local_; } const LocalInfo::LocalInfo& localInfo() override { return *local_info_; } + std::chrono::milliseconds statsFlushInterval() override { return config_->statsFlushInterval(); } + private: void flushStats(); void initialize(Options& options, Network::Address::InstanceConstSharedPtr local_address, diff --git a/test/extensions/stats_sinks/hystrix/BUILD b/test/extensions/stats_sinks/hystrix/BUILD new file mode 100644 index 0000000000000..67a69c5304c4b --- /dev/null +++ b/test/extensions/stats_sinks/hystrix/BUILD @@ -0,0 +1,41 @@ +licenses(["notice"]) # Apache 2 + +load( + "//bazel:envoy_build_system.bzl", + "envoy_package", +) +load( + "//test/extensions:extensions_build_system.bzl", + "envoy_extension_cc_test", +) + +envoy_package() + +envoy_extension_cc_test( + name = "config_test", + srcs = ["config_test.cc"], + extension_name = "envoy.stat_sinks.hystrix", + deps = [ + "//include/envoy/registry", + "//source/common/protobuf:utility_lib", + "//source/extensions/stat_sinks/hystrix:config", + "//test/mocks/server:server_mocks", + "//test/test_common:environment_lib", + "//test/test_common:network_utility_lib", + "//test/test_common:utility_lib", + ], +) + +envoy_extension_cc_test( + name = "hystrix_test", + srcs = ["hystrix_test.cc"], + extension_name = "envoy.stat_sinks.hystrix", + deps = [ + "//source/common/stats:stats_lib", + "//source/extensions/stat_sinks/hystrix:hystrix_lib", + "//test/mocks/server:server_mocks", + "//test/mocks/stats:stats_mocks", + "//test/mocks/upstream:upstream_mocks", + "//test/test_common:utility_lib", + ], +) diff --git a/test/extensions/stats_sinks/hystrix/config_test.cc b/test/extensions/stats_sinks/hystrix/config_test.cc new file mode 100644 index 0000000000000..520b36ae51a28 --- /dev/null +++ b/test/extensions/stats_sinks/hystrix/config_test.cc @@ -0,0 +1,49 @@ +#include "envoy/config/bootstrap/v2/bootstrap.pb.h" +#include "envoy/registry/registry.h" + +#include "common/protobuf/utility.h" + +#include "extensions/stat_sinks/hystrix/config.h" +#include "extensions/stat_sinks/hystrix/hystrix.h" +#include "extensions/stat_sinks/well_known_names.h" + +#include "test/mocks/server/mocks.h" +#include "test/test_common/environment.h" +#include "test/test_common/network_utility.h" +#include "test/test_common/utility.h" + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +using testing::NiceMock; +using testing::Return; +using testing::ReturnRef; +using testing::_; + +namespace Envoy { +namespace Extensions { +namespace StatSinks { +namespace Hystrix { + +TEST(StatsConfigTest, ValidHystrixSink) { + const std::string name = StatsSinkNames::get().HYSTRIX; + + envoy::config::metrics::v2::HystrixSink sink_config; + + Server::Configuration::StatsSinkFactory* factory = + Registry::FactoryRegistry::getFactory(name); + ASSERT_NE(factory, nullptr); + + ProtobufTypes::MessagePtr message = factory->createEmptyConfigProto(); + MessageUtil::jsonConvert(sink_config, *message); + + NiceMock server; + Stats::SinkPtr sink = factory->createStatsSink(*message, server); + EXPECT_NE(sink, nullptr); + EXPECT_NE(dynamic_cast(sink.get()), nullptr); +} + +} // namespace Hystrix +} // namespace StatSinks +} // namespace Extensions +} // namespace Envoy diff --git a/test/extensions/stats_sinks/hystrix/hystrix_test.cc b/test/extensions/stats_sinks/hystrix/hystrix_test.cc new file mode 100644 index 0000000000000..0226d4f6f8dc1 --- /dev/null +++ b/test/extensions/stats_sinks/hystrix/hystrix_test.cc @@ -0,0 +1,191 @@ +#include +#include +#include + +#include "common/stats/stats_impl.h" + +#include "extensions/stat_sinks/hystrix/hystrix.h" + +#include "test/mocks/server/mocks.h" +#include "test/mocks/stats/mocks.h" +#include "test/mocks/upstream/mocks.h" +#include "test/test_common/utility.h" + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +using testing::InSequence; +using testing::Invoke; +using testing::NiceMock; +using testing::Return; +using testing::_; + +namespace Envoy { +namespace Extensions { +namespace StatSinks { +namespace Hystrix { + +class HystrixSinkTest : public testing::Test { +public: + HystrixSinkTest() { sink_.reset(new HystrixSink(server_, 10)); } + + absl::string_view getStreamField(absl::string_view dataMessage, absl::string_view key) { + absl::string_view::size_type key_pos = dataMessage.find(key); + EXPECT_NE(absl::string_view::npos, key_pos); + absl::string_view trimDataBeforeKey = dataMessage.substr(key_pos); + key_pos = trimDataBeforeKey.find(" "); + EXPECT_NE(absl::string_view::npos, key_pos); + absl::string_view trimDataAfterValue = trimDataBeforeKey.substr(key_pos + 1); + key_pos = trimDataAfterValue.find(","); + EXPECT_NE(absl::string_view::npos, key_pos); + absl::string_view actual = trimDataAfterValue.substr(0, key_pos); + return actual; + } + + Buffer::OwnedImpl createClusterAndCallbacks() { + + // set cluster + cluster_.info_->name_ = "test_cluster"; + cluster_map_.emplace("test_cluster", cluster_); + ON_CALL(server_, clusterManager()).WillByDefault(ReturnRef(cluster_manager_)); + ON_CALL(cluster_manager_, clusters()).WillByDefault(Return(cluster_map_)); + + // set callbacks to send data to buffer + Buffer::OwnedImpl buffer; + auto encode_callback = [&buffer](Buffer::Instance& data, bool) { + buffer.add( + data); // This will append to the end of the buffer, so multiple calls will all be dumped + // one after another into this buffer. See Buffer::Instance for other buffer + // buffer modification options. + }; + ON_CALL(callbacks_, encodeData(_, _)).WillByDefault(Invoke(encode_callback)); + + return buffer; + } + + NiceMock callbacks_; + NiceMock server_; + NiceMock cluster_; + Upstream::ClusterManager::ClusterInfoMap cluster_map_; + + NiceMock cluster_manager_; + std::unique_ptr sink_; +}; + +TEST_F(HystrixSinkTest, EmptyFlush) { + InSequence s; + Buffer::OwnedImpl buffer = createClusterAndCallbacks(); + // register callback to sink + sink_->registerConnection(&callbacks_); + + sink_->beginFlush(); + sink_->endFlush(); + std::string data_message = TestUtility::bufferToString(buffer); + EXPECT_EQ(getStreamField(data_message, "errorPercentage"), "0"); + EXPECT_EQ(getStreamField(data_message, "errorCount"), "0"); + EXPECT_EQ(getStreamField(data_message, "requestCount"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSemaphoreRejected"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSuccess"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountTimeout"), "0"); +} + +TEST_F(HystrixSinkTest, BasicFlow) { + InSequence s; + + Buffer::OwnedImpl buffer = createClusterAndCallbacks(); + // register callback to sink + sink_->registerConnection(&callbacks_); + + NiceMock success_counter; + success_counter.name_ = "cluster.test_cluster.upstream_rq_2xx"; + NiceMock error_counter; + error_counter.name_ = "cluster.test_cluster.upstream_rq_5xx"; + NiceMock timeout_counter; + timeout_counter.name_ = "cluster.test_cluster.upstream_rq_timeout"; + NiceMock rejected_counter; + rejected_counter.name_ = "cluster.test_cluster.upstream_rq_pending_overflow"; + + for (int i = 0; i < 12; i++) { + buffer.drain(buffer.length()); + ON_CALL(timeout_counter, value()).WillByDefault(Return((i + 1) * 3)); + ON_CALL(error_counter, value()).WillByDefault(Return((i + 1) * 17)); + ON_CALL(success_counter, value()).WillByDefault(Return((i + 1) * 7)); + ON_CALL(rejected_counter, value()).WillByDefault(Return((i + 1) * 8)); + sink_->beginFlush(); + sink_->flushCounter(timeout_counter, 1); + sink_->flushCounter(error_counter, 1); + sink_->flushCounter(success_counter, 1); + sink_->flushCounter(rejected_counter, 1); + sink_->endFlush(); + } + + // //std::string rolling_map = sink_->getStats().printRollingWindow(); + std::string rolling_map = sink_->getStats().printRollingWindow(); + std::size_t pos = rolling_map.find("cluster.test_cluster.total"); + EXPECT_NE(std::string::npos, pos); + // //EXPECT_NE(absl::string_view::npos, map.find("cluster.test_cluster.total")); + + std::string data_message = TestUtility::bufferToString(buffer); + + // check stream format and data + EXPECT_EQ(getStreamField(data_message, "errorCount"), "140"); // note that on regular operation, + // 5xx and timeout are raised + // together, so timeouts are reduced + // from 5xx count + EXPECT_EQ(getStreamField(data_message, "requestCount"), "320"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSemaphoreRejected"), "80"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSuccess"), "70"); + EXPECT_EQ(getStreamField(data_message, "rollingCountTimeout"), "30"); + EXPECT_EQ(getStreamField(data_message, "errorPercentage"), "78"); + + // check the values are reset + buffer.drain(buffer.length()); + sink_->getStats().resetRollingWindow(); + sink_->beginFlush(); + sink_->endFlush(); + data_message = TestUtility::bufferToString(buffer); + EXPECT_EQ(getStreamField(data_message, "errorPercentage"), "0"); + EXPECT_EQ(getStreamField(data_message, "errorCount"), "0"); + EXPECT_EQ(getStreamField(data_message, "requestCount"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSemaphoreRejected"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountSuccess"), "0"); + EXPECT_EQ(getStreamField(data_message, "rollingCountTimeout"), "0"); +} + +TEST_F(HystrixSinkTest, Disconnect) { + InSequence s; + + Buffer::OwnedImpl buffer = createClusterAndCallbacks(); + + // flush with no connection + NiceMock success_counter; + success_counter.name_ = "cluster.test_cluster.upstream_rq_2xx"; + ON_CALL(success_counter, value()).WillByDefault(Return(1234)); + + sink_->beginFlush(); + sink_->flushCounter(success_counter, 1); + sink_->endFlush(); + EXPECT_EQ(buffer.length(), 0); + + // register callback to sink + sink_->registerConnection(&callbacks_); + sink_->beginFlush(); + sink_->flushCounter(success_counter, 1); + sink_->endFlush(); + std::string data_message = TestUtility::bufferToString(buffer); + EXPECT_EQ(getStreamField(data_message, "rollingCountSuccess"), "0"); + EXPECT_NE(buffer.length(), 0); + + // disconnect + buffer.drain(buffer.length()); + sink_->unregisterConnection(&callbacks_); + sink_->beginFlush(); + sink_->flushCounter(success_counter, 1); + sink_->endFlush(); + EXPECT_EQ(buffer.length(), 0); +} + +} // namespace Hystrix +} // namespace StatSinks +} // namespace Extensions +} // namespace Envoy diff --git a/test/integration/integration_admin_test.cc b/test/integration/integration_admin_test.cc index 164526729faf3..86feba8a840ec 100644 --- a/test/integration/integration_admin_test.cc +++ b/test/integration/integration_admin_test.cc @@ -264,6 +264,13 @@ TEST_P(IntegrationAdminTest, Admin) { EXPECT_EQ(listener_it->get().socket().localAddress()->asString(), (*listener_info_it)->asString()); } + + // TODO (@trabetti) : how to test the endpoint? we don't have response->complete(). + // response = IntegrationUtil::makeSingleRequest(lookupPort("http"), "GET", + // "/hystrix_event_stream", "", + // downstreamProtocol(), version_); + // //EXPECT_TRUE(response->complete()); + // EXPECT_STREQ("503", response->headers().Status()->value().c_str()); } // Successful call to startProfiler requires tcmalloc. diff --git a/test/mocks/server/BUILD b/test/mocks/server/BUILD index 8d97ee9460729..efa4cfa7ee221 100644 --- a/test/mocks/server/BUILD +++ b/test/mocks/server/BUILD @@ -13,6 +13,7 @@ envoy_cc_mock( srcs = ["mocks.cc"], hdrs = ["mocks.h"], deps = [ + "//include/envoy/http:filter_interface", "//include/envoy/server:admin_interface", "//include/envoy/server:configuration_interface", "//include/envoy/server:drain_manager_interface", diff --git a/test/mocks/server/mocks.h b/test/mocks/server/mocks.h index 7cd9680c43bc3..4f063e4aebc64 100644 --- a/test/mocks/server/mocks.h +++ b/test/mocks/server/mocks.h @@ -5,6 +5,7 @@ #include #include +#include "envoy/http/filter.h" #include "envoy/server/admin.h" #include "envoy/server/configuration.h" #include "envoy/server/drain_manager.h" @@ -286,6 +287,7 @@ class MockInstance : public Instance { MOCK_METHOD0(httpTracer, Tracing::HttpTracer&()); MOCK_METHOD0(threadLocal, ThreadLocal::Instance&()); MOCK_METHOD0(localInfo, const LocalInfo::LocalInfo&()); + MOCK_METHOD0(statsFlushInterval, std::chrono::milliseconds()); testing::NiceMock thread_local_; Stats::IsolatedStoreImpl stats_store_; diff --git a/test/server/http/admin_test.cc b/test/server/http/admin_test.cc index a9ce5d3a3e918..af438fbd29c62 100644 --- a/test/server/http/admin_test.cc +++ b/test/server/http/admin_test.cc @@ -81,7 +81,7 @@ class AdminInstanceTest : public testing::TestWithParam Http::Code { - return Http::Code::Accepted; - }; + auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&, + AdminFilter&) -> Http::Code { return Http::Code::Accepted; }; // Test removable handler. EXPECT_NO_LOGS(EXPECT_TRUE(admin_.addHandler("/foo/bar", "hello", callback, true, false))); @@ -207,9 +209,8 @@ TEST_P(AdminInstanceTest, CustomHandler) { } TEST_P(AdminInstanceTest, RejectHandlerWithXss) { - auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&) -> Http::Code { - return Http::Code::Accepted; - }; + auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&, + AdminFilter&) -> Http::Code { return Http::Code::Accepted; }; EXPECT_LOG_CONTAINS("error", "filter \"/foo\" contains invalid character '<'", EXPECT_FALSE(admin_.addHandler("/foo", "hello", @@ -217,9 +218,8 @@ TEST_P(AdminInstanceTest, RejectHandlerWithXss) { } TEST_P(AdminInstanceTest, RejectHandlerWithEmbeddedQuery) { - auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&) -> Http::Code { - return Http::Code::Accepted; - }; + auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&, + AdminFilter&) -> Http::Code { return Http::Code::Accepted; }; EXPECT_LOG_CONTAINS("error", "filter \"/bar?queryShouldNotBeInPrefix\" contains invalid character '?'", EXPECT_FALSE(admin_.addHandler("/bar?queryShouldNotBeInPrefix", "hello", @@ -227,9 +227,8 @@ TEST_P(AdminInstanceTest, RejectHandlerWithEmbeddedQuery) { } TEST_P(AdminInstanceTest, EscapeHelpTextWithPunctuation) { - auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&) -> Http::Code { - return Http::Code::Accepted; - }; + auto callback = [](absl::string_view, Http::HeaderMap&, Buffer::Instance&, + AdminFilter&) -> Http::Code { return Http::Code::Accepted; }; // It's OK to have help text with HTML characters in it, but when we render the home // page they need to be escaped.