diff --git a/include/envoy/common/time.h b/include/envoy/common/time.h index 303b795d88756..a7d2bcb9f1e4a 100644 --- a/include/envoy/common/time.h +++ b/include/envoy/common/time.h @@ -1,6 +1,21 @@ #pragma once +#include "envoy/common/pure.h" + /** * Less typing for common system time type. */ typedef std::chrono::time_point SystemTime; + +/** + * Abstraction for getting the current system time. Useful for testing. + */ +class SystemTimeSource { +public: + virtual ~SystemTimeSource() {} + + /** + * @return the current system time. + */ + virtual SystemTime currentSystemTime() PURE; +}; diff --git a/source/common/upstream/cluster_manager_impl.cc b/source/common/upstream/cluster_manager_impl.cc index cfa0cefb5465f..5573efee132c5 100644 --- a/source/common/upstream/cluster_manager_impl.cc +++ b/source/common/upstream/cluster_manager_impl.cc @@ -141,7 +141,7 @@ void ClusterManagerImpl::loadCluster(const Json::Object& cluster, Stats::Store& } new_cluster->setOutlierDetector(OutlierDetectorImplFactory::createForCluster( - *new_cluster, cluster, dns_resolver.dispatcher())); + *new_cluster, cluster, dns_resolver.dispatcher(), runtime, stats)); primary_clusters_.emplace(new_cluster->name(), new_cluster); } diff --git a/source/common/upstream/outlier_detection_impl.cc b/source/common/upstream/outlier_detection_impl.cc index fd68de30a0221..64f15d0561743 100644 --- a/source/common/upstream/outlier_detection_impl.cc +++ b/source/common/upstream/outlier_detection_impl.cc @@ -1,22 +1,50 @@ #include "outlier_detection_impl.h" +#include "envoy/event/dispatcher.h" + #include "common/common/assert.h" +#include "common/http/codes.h" namespace Upstream { OutlierDetectorPtr OutlierDetectorImplFactory::createForCluster(Cluster& cluster, const Json::Object& cluster_config, - Event::Dispatcher& dispatcher) { + Event::Dispatcher& dispatcher, + Runtime::Loader& runtime, + Stats::Store& stats) { // Right now we don't support any configuration but in order to make the config backwards // compatible we just look for an empty object. if (cluster_config.hasObject("outlier_detection")) { - return OutlierDetectorPtr{new OutlierDetectorImpl(cluster, dispatcher)}; + return OutlierDetectorPtr{new ProdOutlierDetectorImpl(cluster, dispatcher, runtime, stats)}; } else { return nullptr; } } -OutlierDetectorImpl::OutlierDetectorImpl(Cluster& cluster, Event::Dispatcher&) { +void OutlierDetectorHostSinkImpl::eject(SystemTime ejection_time) { + ASSERT(!host_.lock()->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); + host_.lock()->healthFlagSet(Host::HealthFlag::FAILED_OUTLIER_CHECK); + num_ejections_++; + ejection_time_ = ejection_time; +} + +void OutlierDetectorHostSinkImpl::putHttpResponseCode(uint64_t response_code) { + if (Http::CodeUtility::is5xx(response_code)) { + if (++consecutive_5xx_ == + detector_.runtime().snapshot().getInteger("outlier_detection.consecutive_5xx", 5)) { + detector_.onConsecutive5xx(host_.lock()); + } + } else { + consecutive_5xx_ = 0; + } +} + +OutlierDetectorImpl::OutlierDetectorImpl(Cluster& cluster, Event::Dispatcher& dispatcher, + Runtime::Loader& runtime, Stats::Store& stats, + SystemTimeSource& time_source) + : dispatcher_(dispatcher), runtime_(runtime), time_source_(time_source), + stats_(generateStats(cluster.name(), stats)), + interval_timer_(dispatcher.createTimer([this]() -> void { onIntervalTimer(); })) { for (HostPtr host : cluster.hosts()) { addHostSink(host); } @@ -29,16 +57,101 @@ OutlierDetectorImpl::OutlierDetectorImpl(Cluster& cluster, Event::Dispatcher&) { for (HostPtr host : hosts_removed) { ASSERT(host_sinks_.count(host) == 1); + if (host->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)) { + ASSERT(stats_.ejections_active_.value() > 0); + stats_.ejections_active_.dec(); + } + host_sinks_.erase(host); } }); + + armIntervalTimer(); } void OutlierDetectorImpl::addHostSink(HostPtr host) { ASSERT(host_sinks_.count(host) == 0); - OutlierDetectorHostSinkImpl* sink = new OutlierDetectorHostSinkImpl(); + OutlierDetectorHostSinkImpl* sink = new OutlierDetectorHostSinkImpl(*this, host); host_sinks_[host] = sink; host->setOutlierDetector(OutlierDetectorHostSinkPtr{sink}); } +void OutlierDetectorImpl::armIntervalTimer() { + interval_timer_->enableTimer(std::chrono::milliseconds( + runtime_.snapshot().getInteger("outlier_detection.interval_ms", 10000))); +} + +void OutlierDetectorImpl::checkHostForUneject(HostPtr host, OutlierDetectorHostSinkImpl* sink, + SystemTime now) { + if (!host->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)) { + return; + } + + std::chrono::milliseconds base_eject_time = std::chrono::milliseconds( + runtime_.snapshot().getInteger("outlier_detection.base_ejection_time_ms", 30000)); + ASSERT(sink->numEjections() > 0) + if ((base_eject_time * sink->numEjections()) <= (now - sink->ejectionTime())) { + stats_.ejections_active_.dec(); + host->healthFlagClear(Host::HealthFlag::FAILED_OUTLIER_CHECK); + runCallbacks(host); + } +} + +void OutlierDetectorImpl::ejectHost(HostPtr host) { + uint64_t max_ejection_percent = + std::min(100UL, runtime_.snapshot().getInteger("outlier_detection.max_ejection_percent", 10)); + if ((stats_.ejections_active_.value() / host_sinks_.size()) < max_ejection_percent) { + stats_.ejections_total_.inc(); + if (runtime_.snapshot().featureEnabled("outlier_detection.enforcing", 100)) { + stats_.ejections_active_.inc(); + host_sinks_[host]->eject(time_source_.currentSystemTime()); + runCallbacks(host); + } + } else { + stats_.ejections_overflow_.inc(); + } +} + +OutlierDetectionStats OutlierDetectorImpl::generateStats(const std::string& name, + Stats::Store& store) { + std::string prefix(fmt::format("cluster.{}.outlier_detection.", name)); + return {ALL_OUTLIER_DETECTION_STATS(POOL_COUNTER_PREFIX(store, prefix), + POOL_GAUGE_PREFIX(store, prefix))}; +} + +void OutlierDetectorImpl::onConsecutive5xx(HostPtr host) { + // This event will come from all threads, so we synchronize with a post to the main thread. + dispatcher_.post([this, host]() -> void { onConsecutive5xxWorker(host); }); +} + +void OutlierDetectorImpl::onConsecutive5xxWorker(HostPtr host) { + // This comes in cross thread. There is a chance that the host has already been removed from + // the set. If so, just ignore it. + if (host_sinks_.count(host) == 0) { + return; + } + + if (host->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)) { + return; + } + + stats_.ejections_consecutive_5xx_.inc(); + ejectHost(host); +} + +void OutlierDetectorImpl::onIntervalTimer() { + SystemTime now = time_source_.currentSystemTime(); + for (auto host : host_sinks_) { + checkHostForUneject(host.first, host.second, now); + } + + armIntervalTimer(); +} + +void OutlierDetectorImpl::runCallbacks(HostPtr host) { + for (ChangeStateCb cb : callbacks_) { + cb(host); + } +} + } // Upstream diff --git a/source/common/upstream/outlier_detection_impl.h b/source/common/upstream/outlier_detection_impl.h index 5219fc0b89753..00c70d38b7cff 100644 --- a/source/common/upstream/outlier_detection_impl.h +++ b/source/common/upstream/outlier_detection_impl.h @@ -1,5 +1,8 @@ #pragma once +#include "envoy/common/time.h" +#include "envoy/event/timer.h" +#include "envoy/runtime/runtime.h" #include "envoy/upstream/outlier_detection.h" #include "envoy/upstream/upstream.h" @@ -23,17 +26,52 @@ class OutlierDetectorHostSinkNullImpl : public OutlierDetectorHostSink { class OutlierDetectorImplFactory { public: static OutlierDetectorPtr createForCluster(Cluster& cluster, const Json::Object& cluster_config, - Event::Dispatcher& dispatcher); + Event::Dispatcher& dispatcher, + Runtime::Loader& runtime, Stats::Store& stats); }; +class OutlierDetectorImpl; + /** * Implementation of OutlierDetectorHostSink for the generic detector. */ class OutlierDetectorHostSinkImpl : public OutlierDetectorHostSink { public: + OutlierDetectorHostSinkImpl(OutlierDetectorImpl& detector, HostPtr host) + : detector_(detector), host_(host) {} + + void eject(SystemTime ejection_time); + SystemTime ejectionTime() { return ejection_time_; } + uint32_t numEjections() { return num_ejections_; } + // Upstream::OutlierDetectorHostSink - void putHttpResponseCode(uint64_t) override {} + void putHttpResponseCode(uint64_t response_code) override; void putResponseTime(std::chrono::milliseconds) override {} + +private: + OutlierDetectorImpl& detector_; + std::weak_ptr host_; + std::atomic consecutive_5xx_{0}; + SystemTime ejection_time_; + uint32_t num_ejections_{}; +}; + +/** + * All outlier detection stats. @see stats_macros.h + */ +// clang-format off +#define ALL_OUTLIER_DETECTION_STATS(COUNTER, GAUGE) \ + COUNTER(ejections_total) \ + GAUGE (ejections_active) \ + COUNTER(ejections_overflow) \ + COUNTER(ejections_consecutive_5xx) +// clang-format on + +/** + * Struct definition for all outlier detection stats. @see stats_macros.h + */ +struct OutlierDetectionStats { + ALL_OUTLIER_DETECTION_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT) }; /** @@ -43,16 +81,43 @@ class OutlierDetectorHostSinkImpl : public OutlierDetectorHostSink { */ class OutlierDetectorImpl : public OutlierDetector { public: - OutlierDetectorImpl(Cluster& cluster, Event::Dispatcher& dispatcher); + void onConsecutive5xx(HostPtr host); + Runtime::Loader& runtime() { return runtime_; } // Upstream::OutlierDetector void addChangedStateCb(ChangeStateCb cb) override { callbacks_.push_back(cb); } +protected: + OutlierDetectorImpl(Cluster& cluster, Event::Dispatcher& dispatcher, Runtime::Loader& runtime, + Stats::Store& stats, SystemTimeSource& time_source); + private: void addHostSink(HostPtr host); + void armIntervalTimer(); + void checkHostForUneject(HostPtr host, OutlierDetectorHostSinkImpl* sink, SystemTime now); + void ejectHost(HostPtr host); + static OutlierDetectionStats generateStats(const std::string& name, Stats::Store& store); + void onConsecutive5xxWorker(HostPtr host); + void onIntervalTimer(); + void runCallbacks(HostPtr host); + Event::Dispatcher& dispatcher_; + Runtime::Loader& runtime_; + SystemTimeSource& time_source_; + OutlierDetectionStats stats_; + Event::TimerPtr interval_timer_; std::list callbacks_; std::unordered_map host_sinks_; }; +class ProdOutlierDetectorImpl : public OutlierDetectorImpl, public SystemTimeSource { +public: + ProdOutlierDetectorImpl(Cluster& cluster, Event::Dispatcher& dispatcher, Runtime::Loader& runtime, + Stats::Store& stats) + : OutlierDetectorImpl(cluster, dispatcher, runtime, stats, *this) {} + + // SystemTimeSource + SystemTime currentSystemTime() override { return std::chrono::system_clock::now(); } +}; + } // Upstream diff --git a/test/common/upstream/outlier_detection_impl_test.cc b/test/common/upstream/outlier_detection_impl_test.cc index 9d3c2afed6dbb..82b4756ca0182 100644 --- a/test/common/upstream/outlier_detection_impl_test.cc +++ b/test/common/upstream/outlier_detection_impl_test.cc @@ -2,18 +2,24 @@ #include "common/upstream/upstream_impl.h" #include "test/mocks/event/mocks.h" +#include "test/mocks/runtime/mocks.h" #include "test/mocks/upstream/mocks.h" using testing::_; using testing::NiceMock; +using testing::Return; +using testing::SaveArg; namespace Upstream { TEST(OutlierDetectorImplFactoryTest, NoDetector) { Json::ObjectPtr loader = Json::Factory::LoadFromString("{}"); - MockCluster cluster; - Event::MockDispatcher dispatcher; - EXPECT_EQ(nullptr, OutlierDetectorImplFactory::createForCluster(cluster, *loader, dispatcher)); + NiceMock cluster; + NiceMock dispatcher; + NiceMock runtime; + Stats::IsolatedStoreImpl stats_store; + EXPECT_EQ(nullptr, OutlierDetectorImplFactory::createForCluster(cluster, *loader, dispatcher, + runtime, stats_store)); } TEST(OutlierDetectorImplFactoryTest, Detector) { @@ -26,28 +32,221 @@ TEST(OutlierDetectorImplFactoryTest, Detector) { Json::ObjectPtr loader = Json::Factory::LoadFromString(json); NiceMock cluster; NiceMock dispatcher; - EXPECT_NE(nullptr, OutlierDetectorImplFactory::createForCluster(cluster, *loader, dispatcher)); + NiceMock runtime; + Stats::IsolatedStoreImpl stats_store; + EXPECT_NE(nullptr, OutlierDetectorImplFactory::createForCluster(cluster, *loader, dispatcher, + runtime, stats_store)); } -TEST(OutlierDetectorImplTest, Callbacks) { - NiceMock cluster; - Event::MockDispatcher dispatcher; +class TestOutlierDetectorImpl : public OutlierDetectorImpl, public SystemTimeSource { +public: + TestOutlierDetectorImpl(Cluster& cluster, Event::Dispatcher& dispatcher, Runtime::Loader& runtime, + Stats::Store& stats) + : OutlierDetectorImpl(cluster, dispatcher, runtime, stats, *this) {} + + // SystemTimeSource + MOCK_METHOD0(currentSystemTime, SystemTime()); +}; + +class CallbackChecker { +public: + MOCK_METHOD1(check, void(HostPtr host)); +}; + +class OutlierDetectorImplTest : public testing::Test { +public: + OutlierDetectorImplTest() { + ON_CALL(runtime_.snapshot_, featureEnabled("outlier_detection.enforcing", 100)) + .WillByDefault(Return(true)); + } + + NiceMock cluster_; + NiceMock dispatcher_; + NiceMock runtime_; + Event::MockTimer* interval_timer_ = new Event::MockTimer(&dispatcher_); + Stats::IsolatedStoreImpl stats_store_; + CallbackChecker checker_; +}; + +TEST_F(OutlierDetectorImplTest, BasicFlow) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); + + cluster_.hosts_.push_back(HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:81", false, 1, "")}); + cluster_.runCallbacks({cluster_.hosts_[1]}, {}); + + // Cause a consecutive 5xx error. + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(200); + cluster_.hosts_[0]->outlierDetector().putResponseTime(std::chrono::milliseconds(5)); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(0)))); + EXPECT_CALL(checker_, check(cluster_.hosts_[0])); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + EXPECT_TRUE(cluster_.hosts_[0]->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); + + EXPECT_EQ(1UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); + + // Interval that doesn't bring the host back in. + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(9999)))); + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + interval_timer_->callback_(); + + // Interval that does bring the host back in. + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(30001)))); + EXPECT_CALL(checker_, check(cluster_.hosts_[0])); + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + interval_timer_->callback_(); + EXPECT_FALSE(cluster_.hosts_[0]->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); + + cluster_.runCallbacks({}, cluster_.hosts_); + + EXPECT_EQ(0UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); + EXPECT_EQ(1UL, + stats_store_.counter("cluster.fake_cluster.outlier_detection.ejections_total").value()); + EXPECT_EQ(1UL, + stats_store_.counter("cluster.fake_cluster.outlier_detection.ejections_consecutive_5xx") + .value()); +} + +TEST_F(OutlierDetectorImplTest, RemoveWhileEjected) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); + + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(0)))); + EXPECT_CALL(checker_, check(cluster_.hosts_[0])); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + EXPECT_TRUE(cluster_.hosts_[0]->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); - EXPECT_CALL(cluster, addMemberUpdateCb(_)); - cluster.hosts_ = {HostPtr{new HostImpl(cluster, "tcp://127.0.0.1:80", false, 1, "")}}; - OutlierDetectorImpl detector(cluster, dispatcher); + EXPECT_EQ(1UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); + + std::vector old_hosts = std::move(cluster_.hosts_); + cluster_.runCallbacks({}, old_hosts); + + EXPECT_EQ(0UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); + + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(9999)))); + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + interval_timer_->callback_(); +} + +TEST_F(OutlierDetectorImplTest, Overflow) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); + + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + ON_CALL(runtime_.snapshot_, getInteger("outlier_detection.max_ejection_percent", _)) + .WillByDefault(Return(0)); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + EXPECT_FALSE(cluster_.hosts_[0]->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); + + EXPECT_EQ(0UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); + EXPECT_EQ(1UL, stats_store_.counter("cluster.fake_cluster.outlier_detection.ejections_overflow") + .value()); +} + +TEST_F(OutlierDetectorImplTest, CrossThreadRemoveRace) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); + + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + Event::PostCb post_cb; + EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb)); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + // Remove before the cross thread event comes in. + std::vector old_hosts = std::move(cluster_.hosts_); + cluster_.runCallbacks({}, old_hosts); + post_cb(); + + EXPECT_EQ(0UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); +} + +TEST_F(OutlierDetectorImplTest, CrossThreadFailRace) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); + + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + Event::PostCb post_cb; + EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb)); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + + // Fake another event coming in which sets failed, then fire the callback. We should not actually + // eject anything. + cluster_.hosts_[0]->healthFlagSet(Host::HealthFlag::FAILED_OUTLIER_CHECK); + post_cb(); + + EXPECT_EQ(0UL, + stats_store_.gauge("cluster.fake_cluster.outlier_detection.ejections_active").value()); +} - // Set up callback. Will replace later with real test when we have real functionality. - detector.addChangedStateCb([](HostPtr) -> void {}); +TEST_F(OutlierDetectorImplTest, Consecutive5xxAlreadyEjected) { + EXPECT_CALL(cluster_, addMemberUpdateCb(_)); + cluster_.hosts_ = {HostPtr{new HostImpl(cluster_, "tcp://127.0.0.1:80", false, 1, "")}}; + EXPECT_CALL(*interval_timer_, enableTimer(std::chrono::milliseconds(10000))); + TestOutlierDetectorImpl detector(cluster_, dispatcher_, runtime_, stats_store_); + detector.addChangedStateCb([&](HostPtr host) -> void { checker_.check(host); }); - cluster.hosts_.push_back(HostPtr{new HostImpl(cluster, "tcp://127.0.0.1:81", false, 1, "")}); - cluster.runCallbacks({cluster.hosts_[1]}, {}); + // Cause a consecutive 5xx error. + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); - // Trivial call through tests to be replaced later with real functionality. - cluster.hosts_[0]->outlierDetector().putHttpResponseCode(200); - cluster.hosts_[0]->outlierDetector().putResponseTime(std::chrono::milliseconds(5)); + EXPECT_CALL(detector, currentSystemTime()) + .WillOnce(Return(SystemTime(std::chrono::milliseconds(0)))); + EXPECT_CALL(checker_, check(cluster_.hosts_[0])); + cluster_.hosts_[0]->outlierDetector().putHttpResponseCode(503); + EXPECT_TRUE(cluster_.hosts_[0]->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)); - cluster.runCallbacks({}, cluster.hosts_); + // Cause another consecutive 5xx error. } } // Upstream diff --git a/test/mocks/event/mocks.cc b/test/mocks/event/mocks.cc index c3b3996f383fd..5f2e858498b05 100644 --- a/test/mocks/event/mocks.cc +++ b/test/mocks/event/mocks.cc @@ -16,6 +16,7 @@ MockDispatcher::MockDispatcher() { ON_CALL(*this, clearDeferredDeleteList()) .WillByDefault(Invoke([this]() -> void { to_delete_.clear(); })); ON_CALL(*this, createTimer_(_)).WillByDefault(ReturnNew>()); + ON_CALL(*this, post(_)).WillByDefault(Invoke([this](PostCb cb) -> void { cb(); })); } MockDispatcher::~MockDispatcher() {}