From a57e03ab1a74b4b995c03c4fa7397f730d5a41b9 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Fri, 20 Aug 2021 19:12:18 +0000 Subject: [PATCH 01/12] Improve cpu and memory usage of stats (counters, gauges and text readouts) sink by: 1. Iterating over stats in the store to create a snapshot (instead of creating a vector by iterating over scopes. 2. Adding an API to filter stats for sinking. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 19 +++ envoy/stats/store.h | 22 +++ source/common/stats/allocator_impl.cc | 57 ++++++++ source/common/stats/allocator_impl.h | 57 ++++++++ source/common/stats/isolated_store_impl.h | 56 +++++++ source/common/stats/thread_local_store.cc | 30 ++++ source/common/stats/thread_local_store.h | 15 ++ source/server/server.cc | 29 ++-- test/common/stats/allocator_impl_test.cc | 138 ++++++++++++++++++ test/integration/server.h | 30 ++++ test/mocks/stats/mocks.h | 10 ++ test/server/BUILD | 19 +++ .../server_stats_flush_benchmark_test.cc | 71 +++++++++ test/server/server_test.cc | 51 +++++++ 14 files changed, 588 insertions(+), 16 deletions(-) create mode 100644 test/server/server_stats_flush_benchmark_test.cc diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index f6181553ad851..ef64aadd42baa 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -58,6 +58,25 @@ class Allocator { virtual const SymbolTable& constSymbolTable() const PURE; virtual SymbolTable& symbolTable() PURE; + /** + * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold + * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. + * @param f_size functor that is provided the number of all sinked stats. + * @param f_stat functor that is provided one sinked stat at a time. + */ + virtual void forEachSinkedCounter(std::function f_size, + std::function f_stat) PURE; + virtual void forEachSinkedGauge(std::function f_size, + std::function f_stat) PURE; + virtual void forEachSinkedTextReadout(std::function f_size, + std::function f_stat) PURE; + + /** + * @param filter should return true if the passed in stat needs to be sinked. + */ + virtual void setCounterSinkFilter(std::function filter) PURE; + virtual void setGaugeSinkFilter(std::function filter) PURE; + virtual void setTextReadoutSinkFilter(std::function filter) PURE; // TODO(jmarantz): create a parallel mechanism to instantiate histograms. At // the moment, histograms don't fit the same pattern of counters and gauges // as they are not actually created in the context of a stats allocator. diff --git a/envoy/stats/store.h b/envoy/stats/store.h index 191ed0f8589c9..49176016c0129 100644 --- a/envoy/stats/store.h +++ b/envoy/stats/store.h @@ -7,6 +7,7 @@ #include "envoy/common/pure.h" #include "envoy/stats/histogram.h" #include "envoy/stats/scope.h" +#include "envoy/stats/stats.h" #include "envoy/stats/stats_matcher.h" #include "envoy/stats/tag_producer.h" @@ -48,6 +49,27 @@ class Store : public Scope { * @return a list of all known histograms. */ virtual std::vector histograms() const PURE; + + /** + * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold + * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. + * @param f_size functor that is provided the number of all sinked stats. + * @param f_stat functor that is provided one sinked stat at a time. + */ + virtual void forEachSinkedCounter(std::function f_size, + std::function f_stat) PURE; + + virtual void forEachSinkedGauge(std::function f_size, + std::function f_stat) PURE; + + virtual void forEachSinkedTextReadout(std::function f_size, + std::function f_stat) PURE; + /** + * @param filter should return true if the passed in stat needs to be sinked. + */ + virtual void setCounterSinkFilter(std::function filter) PURE; + virtual void setGaugeSinkFilter(std::function filter) PURE; + virtual void setTextReadoutSinkFilter(std::function filter) PURE; }; using StorePtr = std::unique_ptr; diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index 3c64b6ba49511..d6c79b0b24209 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -124,6 +124,10 @@ class CounterImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.counters_.erase(statName()); ASSERT(count == 1); + // Erase the counter from sinked_counters_ if a sink filter was provided. + if (alloc_.counter_sink_filter_ != nullptr) { + alloc_.sinked_counters_.erase(this); + } } // Stats::Counter @@ -168,6 +172,10 @@ class GaugeImpl : public StatsSharedImpl { void removeFromSetLockHeld() override ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) { const size_t count = alloc_.gauges_.erase(statName()); ASSERT(count == 1); + // Erase the gauge from sinked_gauges_ if a sink filter was provided. + if (alloc_.gauge_sink_filter_ != nullptr) { + alloc_.sinked_gauges_.erase(this); + } } // Stats::Gauge @@ -240,6 +248,10 @@ class TextReadoutImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.text_readouts_.erase(statName()); ASSERT(count == 1); + // Erase from sinked_text_readouts_ if a sink filter was provided. + if (alloc_.text_readout_sink_filter_ != nullptr) { + alloc_.sinked_text_readouts_.erase(this); + } } // Stats::TextReadout @@ -269,6 +281,9 @@ CounterSharedPtr AllocatorImpl::makeCounter(StatName name, StatName tag_extracte } auto counter = CounterSharedPtr(makeCounterInternal(name, tag_extracted_name, stat_name_tags)); counters_.insert(counter.get()); + if (counter_sink_filter_ && counter_sink_filter_(*counter)) { + sinked_counters_.insert(counter.get()); + } return counter; } @@ -285,6 +300,9 @@ GaugeSharedPtr AllocatorImpl::makeGauge(StatName name, StatName tag_extracted_na auto gauge = GaugeSharedPtr(new GaugeImpl(name, *this, tag_extracted_name, stat_name_tags, import_mode)); gauges_.insert(gauge.get()); + if (gauge_sink_filter_ && gauge_sink_filter_(*gauge)) { + sinked_gauges_.insert(gauge.get()); + } return gauge; } @@ -300,6 +318,9 @@ TextReadoutSharedPtr AllocatorImpl::makeTextReadout(StatName name, StatName tag_ auto text_readout = TextReadoutSharedPtr(new TextReadoutImpl(name, *this, tag_extracted_name, stat_name_tags)); text_readouts_.insert(text_readout.get()); + if (text_readout_sink_filter_ && text_readout_sink_filter_(*text_readout)) { + sinked_text_readouts_.insert(text_readout.get()); + } return text_readout; } @@ -316,5 +337,41 @@ Counter* AllocatorImpl::makeCounterInternal(StatName name, StatName tag_extracte return new CounterImpl(name, *this, tag_extracted_name, stat_name_tags); } +void AllocatorImpl::forEachSinkedCounter(std::function f_size, + std::function f_stat) { + Thread::LockGuard lock(mutex_); + forEachSinkedStat(f_size, f_stat, counter_sink_filter_, counters_, sinked_counters_); +} + +void AllocatorImpl::forEachSinkedGauge(std::function f_size, + std::function f_stat) { + Thread::LockGuard lock(mutex_); + forEachSinkedStat(f_size, f_stat, gauge_sink_filter_, gauges_, sinked_gauges_); +} + +void AllocatorImpl::forEachSinkedTextReadout(std::function f_size, + std::function f_stat) { + Thread::LockGuard lock(mutex_); + forEachSinkedStat(f_size, f_stat, text_readout_sink_filter_, text_readouts_, + sinked_text_readouts_); +} + +void AllocatorImpl::setCounterSinkFilter( + std::function counter_sink_filter) { + ASSERT(counter_sink_filter_ == nullptr); + counter_sink_filter_ = counter_sink_filter; +} + +void AllocatorImpl::setGaugeSinkFilter(std::function gauge_sink_filter) { + ASSERT(gauge_sink_filter_ == nullptr); + gauge_sink_filter_ = gauge_sink_filter; +} + +void AllocatorImpl::setTextReadoutSinkFilter( + std::function text_readout_sink_filter) { + ASSERT(text_readout_sink_filter_ == nullptr); + text_readout_sink_filter_ = text_readout_sink_filter; +} + } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h index db656e2e40134..a8393444e12bd 100644 --- a/source/common/stats/allocator_impl.h +++ b/source/common/stats/allocator_impl.h @@ -33,6 +33,21 @@ class AllocatorImpl : public Allocator { SymbolTable& symbolTable() override { return symbol_table_; } const SymbolTable& constSymbolTable() const override { return symbol_table_; } + void forEachSinkedCounter(std::function, + std::function) override; + + void forEachSinkedGauge(std::function, + std::function) override; + + void forEachSinkedTextReadout(std::function, + std::function) override; + + void setCounterSinkFilter(std::function filter) override; + + void setGaugeSinkFilter(std::function filter) override; + + void setTextReadoutSinkFilter(std::function filter) override; + #ifndef ENVOY_CONFIG_COVERAGE void debugPrint(); #endif @@ -58,14 +73,34 @@ class AllocatorImpl : public Allocator { friend class TextReadoutImpl; friend class NotifyingAllocatorImpl; + // We don't need to check StatName to compare sinked stats. + template + using SinkedStatsSet = absl::flat_hash_set>; + void removeCounterFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); void removeGaugeFromSetLockHeld(Gauge* gauge) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); void removeTextReadoutFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + // Helper method to iterate over sinked stats based on whether or not a stats sink function was + // provided. + template + void forEachSinkedStat(std::function& f_size, + std::function& f_stat, + std::function& sink_filter, + StatSet& stats, SinkedStatsSet& sinked_stats); + StatSet counters_ ABSL_GUARDED_BY(mutex_); StatSet gauges_ ABSL_GUARDED_BY(mutex_); StatSet text_readouts_ ABSL_GUARDED_BY(mutex_); + SinkedStatsSet sinked_counters_ ABSL_GUARDED_BY(mutex_); + SinkedStatsSet sinked_gauges_ ABSL_GUARDED_BY(mutex_); + SinkedStatsSet sinked_text_readouts_ ABSL_GUARDED_BY(mutex_); + + std::function counter_sink_filter_; + std::function gauge_sink_filter_; + std::function text_readout_sink_filter_; + SymbolTable& symbol_table_; // A mutex is needed here to protect both the stats_ object from both @@ -77,5 +112,27 @@ class AllocatorImpl : public Allocator { Thread::ThreadSynchronizer sync_; }; +template +void AllocatorImpl::forEachSinkedStat(std::function& f_size, + std::function& f_stat, + std::function& sink_filter, + StatSet& stats, + SinkedStatsSet& sinked_stats) { + ASSERT(!mutex_.tryLock()); + // Check if a sink filter method was provided. + if (sink_filter != nullptr) { + f_size(sinked_stats.size()); + for (auto& stat : sinked_stats) { + f_stat(*stat); + } + } else { + // Iterate over all stats if a filter method was not provided. + f_size(stats.size()); + for (auto& stat : stats) { + f_stat(*stat); + } + } +} + } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/isolated_store_impl.h b/source/common/stats/isolated_store_impl.h index f0f924ebda541..d5c4c338bc9c7 100644 --- a/source/common/stats/isolated_store_impl.h +++ b/source/common/stats/isolated_store_impl.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "envoy/stats/stats.h" @@ -45,6 +46,9 @@ template class IsolatedStatsCache { RefcountPtr new_stat = counter_alloc_(name); stats_.emplace(new_stat->statName(), new_stat); + if (sink_filter_ && sink_filter_(*new_stat)) { + sinked_stats_.insert(new_stat.get()); + } return *new_stat; } @@ -56,6 +60,9 @@ template class IsolatedStatsCache { RefcountPtr new_stat = gauge_alloc_(name, import_mode); stats_.emplace(new_stat->statName(), new_stat); + if (sink_filter_ && sink_filter_(*new_stat)) { + sinked_stats_.insert(new_stat.get()); + } return *new_stat; } @@ -78,6 +85,9 @@ template class IsolatedStatsCache { RefcountPtr new_stat = text_readout_alloc_(name, type); stats_.emplace(new_stat->statName(), new_stat); + if (sink_filter_ && sink_filter_(*new_stat)) { + sinked_stats_.insert(new_stat.get()); + } return *new_stat; } @@ -100,6 +110,23 @@ template class IsolatedStatsCache { return true; } + void forEachSinkedStat(std::function f_size, + std::function f_stat) { + if (sink_filter_ != nullptr) { + f_size(sinked_stats_.size()); + for (auto& stat : sinked_stats_) { + f_stat(*stat); + } + } else { + f_size(stats_.size()); + for (auto const& stat : stats_) { + f_stat(*stat.second); + } + } + } + + void setSinkFilter(std::function filter) { sink_filter_ = filter; } + private: friend class IsolatedStoreImpl; @@ -112,10 +139,12 @@ template class IsolatedStatsCache { } StatNameHashMap> stats_; + absl::flat_hash_set> sinked_stats_; CounterAllocator counter_alloc_; GaugeAllocator gauge_alloc_; HistogramAllocator histogram_alloc_; TextReadoutAllocator text_readout_alloc_; + std::function sink_filter_; }; class IsolatedStoreImpl : public StoreImpl { @@ -205,6 +234,33 @@ class IsolatedStoreImpl : public StoreImpl { return textReadoutFromStatName(storage.statName()); } + void forEachSinkedCounter(std::function f_size, + std::function f_stat) override { + counters_.forEachSinkedStat(f_size, f_stat); + } + + void forEachSinkedGauge(std::function f_size, + std::function f_stat) override { + gauges_.forEachSinkedStat(f_size, f_stat); + } + + void forEachSinkedTextReadout(std::function f_size, + std::function f_stat) override { + text_readouts_.forEachSinkedStat(f_size, f_stat); + } + + void setCounterSinkFilter(std::function filter) override { + counters_.setSinkFilter(filter); + } + + void setGaugeSinkFilter(std::function filter) override { + gauges_.setSinkFilter(filter); + } + + void setTextReadoutSinkFilter(std::function filter) override { + text_readouts_.setSinkFilter(filter); + } + private: IsolatedStoreImpl(std::unique_ptr&& symbol_table); diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc index 3b29f8b5d4df5..1e90d937b5ee8 100644 --- a/source/common/stats/thread_local_store.cc +++ b/source/common/stats/thread_local_store.cc @@ -975,5 +975,35 @@ bool ParentHistogramImpl::usedLockHeld() const { return false; } +void ThreadLocalStoreImpl::forEachSinkedCounter(std::function f_size, + std::function f_stat) { + Thread::LockGuard lock(lock_); + alloc_.forEachSinkedCounter(f_size, f_stat); +} + +void ThreadLocalStoreImpl::forEachSinkedGauge(std::function f_size, + std::function f_stat) { + Thread::LockGuard lock(lock_); + alloc_.forEachSinkedGauge(f_size, f_stat); +} +void ThreadLocalStoreImpl::forEachSinkedTextReadout( + std::function f_size, std::function f_stat) { + Thread::LockGuard lock(lock_); + alloc_.forEachSinkedTextReadout(f_size, f_stat); +} + +void ThreadLocalStoreImpl::setCounterSinkFilter(std::function filter) { + alloc_.setCounterSinkFilter(filter); +} + +void ThreadLocalStoreImpl::setGaugeSinkFilter(std::function filter) { + alloc_.setGaugeSinkFilter(filter); +} + +void ThreadLocalStoreImpl::setTextReadoutSinkFilter( + std::function filter) { + alloc_.setTextReadoutSinkFilter(filter); +} + } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h index eaf2946fd99f7..18b983ac834dd 100644 --- a/source/common/stats/thread_local_store.h +++ b/source/common/stats/thread_local_store.h @@ -244,6 +244,21 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo std::vector textReadouts() const override; std::vector histograms() const override; + void forEachSinkedCounter(std::function f_size, + std::function f_stat) override; + + void forEachSinkedGauge(std::function f_size, + std::function f_stat) override; + + void forEachSinkedTextReadout(std::function f_size, + std::function f_stat) override; + + void setCounterSinkFilter(std::function) override; + + void setGaugeSinkFilter(std::function) override; + + void setTextReadoutSinkFilter(std::function) override; + // Stats::StoreRoot void addSink(Sink& sink) override { timer_sinks_.push_back(sink); } void setTagProducer(TagProducerPtr&& tag_producer) override { diff --git a/source/server/server.cc b/source/server/server.cc index 872e4d54b2a3a..daf3e6dc5c540 100644 --- a/source/server/server.cc +++ b/source/server/server.cc @@ -20,6 +20,7 @@ #include "envoy/server/bootstrap_extension_config.h" #include "envoy/server/instance.h" #include "envoy/server/options.h" +#include "envoy/stats/stats.h" #include "envoy/upstream/cluster_manager.h" #include "source/common/api/api_impl.h" @@ -165,18 +166,16 @@ void InstanceImpl::failHealthcheck(bool fail) { } MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_source) { - snapped_counters_ = store.counters(); - counters_.reserve(snapped_counters_.size()); - for (const auto& counter : snapped_counters_) { - counters_.push_back({counter->latch(), *counter}); - } + store.forEachSinkedCounter([this](std::size_t size) mutable { counters_.reserve(size); }, + [this](Stats::Counter& counter) mutable { + counters_.push_back({counter.latch(), counter}); + }); - snapped_gauges_ = store.gauges(); - gauges_.reserve(snapped_gauges_.size()); - for (const auto& gauge : snapped_gauges_) { - ASSERT(gauge->importMode() != Stats::Gauge::ImportMode::Uninitialized); - gauges_.push_back(*gauge); - } + store.forEachSinkedGauge([this](std::size_t size) mutable { gauges_.reserve(size); }, + [this](Stats::Gauge& gauge) mutable { + ASSERT(gauge.importMode() != Stats::Gauge::ImportMode::Uninitialized); + gauges_.push_back(gauge); + }); snapped_histograms_ = store.histograms(); histograms_.reserve(snapped_histograms_.size()); @@ -184,11 +183,9 @@ MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_sou histograms_.push_back(*histogram); } - snapped_text_readouts_ = store.textReadouts(); - text_readouts_.reserve(snapped_text_readouts_.size()); - for (const auto& text_readout : snapped_text_readouts_) { - text_readouts_.push_back(*text_readout); - } + store.forEachSinkedTextReadout( + [this](std::size_t size) mutable { text_readouts_.reserve(size); }, + [this](Stats::TextReadout& text_readout) { text_readouts_.push_back(text_readout); }); snapshot_time_ = time_source.systemTime(); } diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index 83fb85d68ea8d..0991f6d94ab23 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -1,5 +1,7 @@ #include +#include "envoy/stats/stats.h" + #include "source/common/stats/allocator_impl.h" #include "test/test_common/logging.h" @@ -125,6 +127,142 @@ TEST_F(AllocatorImplTest, RefCountDecAllocRaceSynchronized) { EXPECT_FALSE(alloc_.isMutexLockedForTest()); } +TEST_F(AllocatorImplTest, ForEachSinkedCounter) { + + StatNameHashSet sinked_stat_names; + std::vector sinked_counters; + std::vector unsinked_counters; + + alloc_.setCounterSinkFilter([&sinked_stat_names](const Stats::Counter& counter) { + if (sinked_stat_names.count(counter.statName()) == 1) { + return true; + } + return false; + }); + + size_t n_stats = 11; + + for (size_t idx = 0; idx < n_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("counter.", idx)); + // sink every 3rd stat + if ((idx + 1) % 3 == 0) { + sinked_stat_names.insert(stat_name); + sinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); + } else { + unsinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); + } + } + + size_t n_sinked_counters = 0; + size_t n_iterations = 0; + alloc_.forEachSinkedCounter([&n_sinked_counters](std::size_t size) { n_sinked_counters = size; }, + [&n_iterations, &sinked_stat_names](Stats::Counter& counter) { + EXPECT_EQ(sinked_stat_names.count(counter.statName()), 1); + ++n_iterations; + }); + EXPECT_EQ(n_sinked_counters, 3); + + // Erase all sinked stats. + sinked_counters.clear(); + n_iterations = 0; + alloc_.forEachSinkedCounter([&n_sinked_counters](std::size_t size) { n_sinked_counters = size; }, + [&n_iterations](Stats::Counter&) { ++n_iterations; }); + EXPECT_EQ(n_sinked_counters, 0); + EXPECT_EQ(n_iterations, 0); +} + +TEST_F(AllocatorImplTest, ForEachSinkedGauge) { + + StatNameHashSet sinked_stat_names; + std::vector sinked_gauges; + std::vector unsinked_gauges; + + alloc_.setGaugeSinkFilter([&sinked_stat_names](const Stats::Gauge& gauge) { + if (sinked_stat_names.count(gauge.statName()) == 1) { + return true; + } + return false; + }); + + size_t n_stats = 11; + + for (size_t idx = 0; idx < n_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("gauge.", idx)); + // sink every 5th stat + if ((idx + 1) % 5 == 0) { + sinked_stat_names.insert(stat_name); + sinked_gauges.emplace_back( + alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); + } else { + unsinked_gauges.emplace_back( + alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); + } + } + + size_t n_sinked_gauges = 0; + size_t n_iterations = 0; + alloc_.forEachSinkedGauge([&n_sinked_gauges](std::size_t size) { n_sinked_gauges = size; }, + [&n_iterations, &sinked_stat_names](Stats::Gauge& gauge) { + EXPECT_EQ(sinked_stat_names.count(gauge.statName()), 1); + ++n_iterations; + }); + EXPECT_EQ(n_sinked_gauges, 2); + + // Erase all sinked stats. + sinked_gauges.clear(); + n_iterations = 0; + alloc_.forEachSinkedGauge([&n_sinked_gauges](std::size_t size) { n_sinked_gauges = size; }, + [&n_iterations](Stats::Gauge&) { ++n_iterations; }); + EXPECT_EQ(n_sinked_gauges, 0); + EXPECT_EQ(n_iterations, 0); +} + +TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) { + + StatNameHashSet sinked_stat_names; + std::vector sinked_text_readouts; + std::vector unsinked_text_readouts; + + alloc_.setTextReadoutSinkFilter([&sinked_stat_names](const Stats::TextReadout& text_readout) { + if (sinked_stat_names.count(text_readout.statName()) == 1) { + return true; + } + return false; + }); + + size_t n_stats = 11; + + for (size_t idx = 0; idx < n_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("text_readout.", idx)); + // sink every 2nd stat + if ((idx + 1) % 2 == 0) { + sinked_stat_names.insert(stat_name); + sinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); + } else { + unsinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); + } + } + + size_t n_sinked_text_readouts = 0; + size_t n_iterations = 0; + alloc_.forEachSinkedTextReadout( + [&n_sinked_text_readouts](std::size_t size) { n_sinked_text_readouts = size; }, + [&n_iterations, &sinked_stat_names](Stats::TextReadout& text_readout) { + EXPECT_EQ(sinked_stat_names.count(text_readout.statName()), 1); + ++n_iterations; + }); + EXPECT_EQ(n_sinked_text_readouts, 5); + + // Erase all sinked stats. + sinked_text_readouts.clear(); + n_iterations = 0; + alloc_.forEachSinkedTextReadout( + [&n_sinked_text_readouts](std::size_t size) { n_sinked_text_readouts = size; }, + [&n_iterations](Stats::TextReadout&) { ++n_iterations; }); + EXPECT_EQ(n_sinked_text_readouts, 0); + EXPECT_EQ(n_iterations, 0); +} + } // namespace } // namespace Stats } // namespace Envoy diff --git a/test/integration/server.h b/test/integration/server.h index 94d5f79504cbe..cfd18f017a4fa 100644 --- a/test/integration/server.h +++ b/test/integration/server.h @@ -281,6 +281,36 @@ class TestIsolatedStoreImpl : public StoreRoot { Thread::LockGuard lock(lock_); return store_.counterFromStatNameWithTags(name, tags); } + void forEachSinkedCounter(std::function f_size, + std::function f_stat) override { + Thread::LockGuard lock(lock_); + store_.forEachSinkedCounter(f_size, f_stat); + } + + void forEachSinkedGauge(std::function f_size, + std::function f_stat) override { + Thread::LockGuard lock(lock_); + store_.forEachSinkedGauge(f_size, f_stat); + } + + void forEachSinkedTextReadout(std::function f_size, + std::function f_stat) override { + Thread::LockGuard lock(lock_); + store_.forEachSinkedTextReadout(f_size, f_stat); + } + + void setCounterSinkFilter(std::function filter) override { + store_.setCounterSinkFilter(filter); + } + + void setGaugeSinkFilter(std::function filter) override { + store_.setGaugeSinkFilter(filter); + } + + void setTextReadoutSinkFilter(std::function filter) override { + store_.setTextReadoutSinkFilter(filter); + } + Counter& counterFromString(const std::string& name) override { Thread::LockGuard lock(lock_); return store_.counterFromString(name); diff --git a/test/mocks/stats/mocks.h b/test/mocks/stats/mocks.h index 3adec4d4a8ae2..5e61f39709cd1 100644 --- a/test/mocks/stats/mocks.h +++ b/test/mocks/stats/mocks.h @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -285,6 +286,15 @@ class MockStore : public TestUtil::TestStore { MOCK_METHOD(Histogram&, histogramFromString, (const std::string& name, Histogram::Unit unit)); MOCK_METHOD(TextReadout&, textReadout, (const std::string&)); MOCK_METHOD(std::vector, text_readouts, (), (const)); + MOCK_METHOD(void, forEachSinkedCounter, + (std::function, std::function)); + MOCK_METHOD(void, forEachSinkedGauge, + (std::function, std::function)); + MOCK_METHOD(void, forEachSinkedTextReadout, + (std::function, std::function)); + MOCK_METHOD(void, setCounterSinkFilter, (std::function)); + MOCK_METHOD(void, setGaugeSinkFilter, (std::function)); + MOCK_METHOD(void, setTextReadoutSinkFilter, (std::function)); MOCK_METHOD(CounterOptConstRef, findCounter, (StatName), (const)); MOCK_METHOD(GaugeOptConstRef, findGauge, (StatName), (const)); diff --git a/test/server/BUILD b/test/server/BUILD index 5e50e3cbacb4d..952131ac020fe 100644 --- a/test/server/BUILD +++ b/test/server/BUILD @@ -499,3 +499,22 @@ envoy_benchmark_test( timeout = "long", benchmark_binary = "filter_chain_benchmark_test", ) + +envoy_cc_benchmark_binary( + name = "server_stats_flush_benchmark", + srcs = ["server_stats_flush_benchmark_test.cc"], + external_deps = [ + "benchmark", + ], + deps = [ + "//envoy/stats:stats_interface", + "//source/common/stats:thread_local_store_lib", + "//source/server:server_lib", + "//test/test_common:simulated_time_system_lib", + ], +) + +envoy_benchmark_test( + name = "server_stats_flush_benchmark_test", + benchmark_binary = "server_stats_flush_benchmark", +) diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc new file mode 100644 index 0000000000000..81534f498ce86 --- /dev/null +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -0,0 +1,71 @@ +#include +#include + +#include "envoy/stats/sink.h" +#include "envoy/stats/stats.h" + +#include "source/common/stats/thread_local_store.h" +#include "source/server/server.h" + +#include "test/mocks/stats/mocks.h" +#include "test/test_common/simulated_time_system.h" + +#include "absl/strings/str_cat.h" +#include "benchmark/benchmark.h" +#include "gmock/gmock.h" + +class StatsSinkFlushSpeedTest { + +public: + StatsSinkFlushSpeedTest(uint64_t n_counters, uint64_t n_gauges, uint64_t n_text_readouts) + : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { + + sinks_.emplace_back(new testing::NiceMock()); + // Create counters + for (uint64_t idx = 0; idx < n_counters; ++idx) { + auto stat_name = pool_.add(absl::StrCat("counter.", idx)); + stats_store_.counterFromStatName(stat_name).inc(); + } + // Create gauges + for (uint64_t idx = 0; idx < n_gauges; ++idx) { + auto stat_name = pool_.add(absl::StrCat("gauge.", idx)); + stats_store_.gaugeFromStatName(stat_name, Envoy::Stats::Gauge::ImportMode::NeverImport) + .set(idx); + } + + // Create text readouts + for (uint64_t idx = 0; idx < n_text_readouts; ++idx) { + auto stat_name = pool_.add(absl::StrCat("text_readout.", idx)); + stats_store_.textReadoutFromStatName(stat_name).set(absl::StrCat("text_readout.", idx)); + } + } + + void test(benchmark::State& state) { + for (auto _ : state) { + UNREFERENCED_PARAMETER(state); + Envoy::Server::InstanceUtil::flushMetricsToSinks(sinks_, stats_store_, time_system_); + } + } + +private: + Envoy::Stats::SymbolTableImpl symbol_table_; + Envoy::Stats::StatNamePool pool_; + Envoy::Stats::AllocatorImpl stats_allocator_; + Envoy::Stats::ThreadLocalStoreImpl stats_store_; + std::list sinks_; + Envoy::Event::SimulatedTimeSystem time_system_; +}; + +static void bmLarge(benchmark::State& state) { + uint64_t n_counters = 1000000, n_gauges = 1000000, n_text_readouts = 1000000; + StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); + speed_test.test(state); +} +BENCHMARK(bmLarge)->Unit(::benchmark::kMillisecond); + +static void bmSmall(benchmark::State& state) { + uint64_t n_counters = 10000, n_gauges = 10000, n_text_readouts = 10000; + StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); + speed_test.test(state); +} +BENCHMARK(bmSmall)->Unit(::benchmark::kMillisecond); diff --git a/test/server/server_test.cc b/test/server/server_test.cc index 8e6acff4fe85d..b57247e67609b 100644 --- a/test/server/server_test.cc +++ b/test/server/server_test.cc @@ -101,6 +101,57 @@ TEST(ServerInstanceUtil, flushHelper) { InstanceUtil::flushMetricsToSinks(sinks, mock_store, time_system); } +TEST(ServerInstanceUtil, sinkFilter) { + InSequence s; + + Stats::TestUtil::TestStore store; + Event::SimulatedTimeSystem time_system; + // Set the sink filters + store.setCounterSinkFilter( + [](const Stats::Counter& counter) { return counter.name().compare("hello") == 0; }); + store.setGaugeSinkFilter( + [](const Stats::Gauge& gauge) { return gauge.name().compare("day") == 0; }); + store.setTextReadoutSinkFilter([](const Stats::TextReadout& text_readout) { + return text_readout.name().compare("text") == 0; + }); + + Stats::Counter& c = store.counter("hello"); + c.inc(); + c.inc(); + store.counter("world").inc(); + + store.gauge("sunny", Stats::Gauge::ImportMode::Accumulate).set(9); + store.gauge("day", Stats::Gauge::ImportMode::Accumulate).set(5); + + store.textReadout("Text").set("don't sink"); + store.textReadout("text").set("is important"); + + std::list sinks; + InstanceUtil::flushMetricsToSinks(sinks, store, time_system); + // Make sure that counters have been latched even if there are no sinks. + EXPECT_EQ(2UL, c.value()); + EXPECT_EQ(0, c.latch()); + + Stats::MockSink* sink = new StrictMock(); + sinks.emplace_back(sink); + EXPECT_CALL(*sink, flush(_)).WillOnce(Invoke([](Stats::MetricSnapshot& snapshot) { + ASSERT_EQ(snapshot.counters().size(), 1); + EXPECT_EQ(snapshot.counters()[0].counter_.get().name(), "hello"); + EXPECT_EQ(snapshot.counters()[0].delta_, 2); + + ASSERT_EQ(snapshot.gauges().size(), 1); + EXPECT_EQ(snapshot.gauges()[0].get().name(), "day"); + EXPECT_EQ(snapshot.gauges()[0].get().value(), 5); + + ASSERT_EQ(snapshot.textReadouts().size(), 1); + EXPECT_EQ(snapshot.textReadouts()[0].get().name(), "text"); + EXPECT_EQ(snapshot.textReadouts()[0].get().value(), "is important"); + })); + c.inc(); + c.inc(); + InstanceUtil::flushMetricsToSinks(sinks, store, time_system); +} + class RunHelperTest : public testing::Test { public: RunHelperTest() { From 9154c943e831aa6cf4933a3f0d8df53493be95f5 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Sat, 21 Aug 2021 00:17:45 +0000 Subject: [PATCH 02/12] fix formatting. Signed-off-by: Pradeep Rao --- .../server_stats_flush_benchmark_test.cc | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index 81534f498ce86..879a5beef7fec 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -14,13 +14,14 @@ #include "benchmark/benchmark.h" #include "gmock/gmock.h" +namespace Envoy { class StatsSinkFlushSpeedTest { public: StatsSinkFlushSpeedTest(uint64_t n_counters, uint64_t n_gauges, uint64_t n_text_readouts) : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { - sinks_.emplace_back(new testing::NiceMock()); + sinks_.emplace_back(new testing::NiceMock()); // Create counters for (uint64_t idx = 0; idx < n_counters; ++idx) { auto stat_name = pool_.add(absl::StrCat("counter.", idx)); @@ -29,8 +30,7 @@ class StatsSinkFlushSpeedTest { // Create gauges for (uint64_t idx = 0; idx < n_gauges; ++idx) { auto stat_name = pool_.add(absl::StrCat("gauge.", idx)); - stats_store_.gaugeFromStatName(stat_name, Envoy::Stats::Gauge::ImportMode::NeverImport) - .set(idx); + stats_store_.gaugeFromStatName(stat_name, Stats::Gauge::ImportMode::NeverImport).set(idx); } // Create text readouts @@ -43,17 +43,17 @@ class StatsSinkFlushSpeedTest { void test(benchmark::State& state) { for (auto _ : state) { UNREFERENCED_PARAMETER(state); - Envoy::Server::InstanceUtil::flushMetricsToSinks(sinks_, stats_store_, time_system_); + Server::InstanceUtil::flushMetricsToSinks(sinks_, stats_store_, time_system_); } } private: - Envoy::Stats::SymbolTableImpl symbol_table_; - Envoy::Stats::StatNamePool pool_; - Envoy::Stats::AllocatorImpl stats_allocator_; - Envoy::Stats::ThreadLocalStoreImpl stats_store_; - std::list sinks_; - Envoy::Event::SimulatedTimeSystem time_system_; + Stats::SymbolTableImpl symbol_table_; + Stats::StatNamePool pool_; + Stats::AllocatorImpl stats_allocator_; + Stats::ThreadLocalStoreImpl stats_store_; + std::list sinks_; + Event::SimulatedTimeSystem time_system_; }; static void bmLarge(benchmark::State& state) { @@ -69,3 +69,5 @@ static void bmSmall(benchmark::State& state) { speed_test.test(state); } BENCHMARK(bmSmall)->Unit(::benchmark::kMillisecond); + +} // namespace Envoy From 83252663b352305d8f8c3784d4e83ab6c5fa18bc Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Mon, 30 Aug 2021 16:26:46 +0000 Subject: [PATCH 03/12] Iterate over stats to be sinked instead of creating and returning a vector of them. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 9 +- envoy/stats/store.h | 6 - source/common/stats/allocator_impl.cc | 54 ++------- source/common/stats/allocator_impl.h | 46 +------- source/common/stats/isolated_store_impl.h | 38 +----- source/common/stats/thread_local_store.cc | 13 -- source/common/stats/thread_local_store.h | 6 - test/common/stats/allocator_impl_test.cc | 138 ---------------------- test/integration/server.h | 12 -- test/mocks/stats/mocks.h | 3 - 10 files changed, 18 insertions(+), 307 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index ef64aadd42baa..d3fab4cd6024f 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -61,7 +61,8 @@ class Allocator { /** * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. - * @param f_size functor that is provided the number of all sinked stats. + * @param f_size functor that is provided the number of all sinked stats. Note this is called + * only once. * @param f_stat functor that is provided one sinked stat at a time. */ virtual void forEachSinkedCounter(std::function f_size, @@ -71,12 +72,6 @@ class Allocator { virtual void forEachSinkedTextReadout(std::function f_size, std::function f_stat) PURE; - /** - * @param filter should return true if the passed in stat needs to be sinked. - */ - virtual void setCounterSinkFilter(std::function filter) PURE; - virtual void setGaugeSinkFilter(std::function filter) PURE; - virtual void setTextReadoutSinkFilter(std::function filter) PURE; // TODO(jmarantz): create a parallel mechanism to instantiate histograms. At // the moment, histograms don't fit the same pattern of counters and gauges // as they are not actually created in the context of a stats allocator. diff --git a/envoy/stats/store.h b/envoy/stats/store.h index 49176016c0129..82cb9e20ba5c3 100644 --- a/envoy/stats/store.h +++ b/envoy/stats/store.h @@ -64,12 +64,6 @@ class Store : public Scope { virtual void forEachSinkedTextReadout(std::function f_size, std::function f_stat) PURE; - /** - * @param filter should return true if the passed in stat needs to be sinked. - */ - virtual void setCounterSinkFilter(std::function filter) PURE; - virtual void setGaugeSinkFilter(std::function filter) PURE; - virtual void setTextReadoutSinkFilter(std::function filter) PURE; }; using StorePtr = std::unique_ptr; diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index d6c79b0b24209..46c97797ee451 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -124,10 +124,6 @@ class CounterImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.counters_.erase(statName()); ASSERT(count == 1); - // Erase the counter from sinked_counters_ if a sink filter was provided. - if (alloc_.counter_sink_filter_ != nullptr) { - alloc_.sinked_counters_.erase(this); - } } // Stats::Counter @@ -172,10 +168,6 @@ class GaugeImpl : public StatsSharedImpl { void removeFromSetLockHeld() override ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) { const size_t count = alloc_.gauges_.erase(statName()); ASSERT(count == 1); - // Erase the gauge from sinked_gauges_ if a sink filter was provided. - if (alloc_.gauge_sink_filter_ != nullptr) { - alloc_.sinked_gauges_.erase(this); - } } // Stats::Gauge @@ -248,10 +240,6 @@ class TextReadoutImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.text_readouts_.erase(statName()); ASSERT(count == 1); - // Erase from sinked_text_readouts_ if a sink filter was provided. - if (alloc_.text_readout_sink_filter_ != nullptr) { - alloc_.sinked_text_readouts_.erase(this); - } } // Stats::TextReadout @@ -281,9 +269,6 @@ CounterSharedPtr AllocatorImpl::makeCounter(StatName name, StatName tag_extracte } auto counter = CounterSharedPtr(makeCounterInternal(name, tag_extracted_name, stat_name_tags)); counters_.insert(counter.get()); - if (counter_sink_filter_ && counter_sink_filter_(*counter)) { - sinked_counters_.insert(counter.get()); - } return counter; } @@ -300,9 +285,6 @@ GaugeSharedPtr AllocatorImpl::makeGauge(StatName name, StatName tag_extracted_na auto gauge = GaugeSharedPtr(new GaugeImpl(name, *this, tag_extracted_name, stat_name_tags, import_mode)); gauges_.insert(gauge.get()); - if (gauge_sink_filter_ && gauge_sink_filter_(*gauge)) { - sinked_gauges_.insert(gauge.get()); - } return gauge; } @@ -318,9 +300,6 @@ TextReadoutSharedPtr AllocatorImpl::makeTextReadout(StatName name, StatName tag_ auto text_readout = TextReadoutSharedPtr(new TextReadoutImpl(name, *this, tag_extracted_name, stat_name_tags)); text_readouts_.insert(text_readout.get()); - if (text_readout_sink_filter_ && text_readout_sink_filter_(*text_readout)) { - sinked_text_readouts_.insert(text_readout.get()); - } return text_readout; } @@ -340,37 +319,28 @@ Counter* AllocatorImpl::makeCounterInternal(StatName name, StatName tag_extracte void AllocatorImpl::forEachSinkedCounter(std::function f_size, std::function f_stat) { Thread::LockGuard lock(mutex_); - forEachSinkedStat(f_size, f_stat, counter_sink_filter_, counters_, sinked_counters_); + f_size(counters_.size()); + for (auto& counter : counters_) { + f_stat(*counter); + } } void AllocatorImpl::forEachSinkedGauge(std::function f_size, std::function f_stat) { Thread::LockGuard lock(mutex_); - forEachSinkedStat(f_size, f_stat, gauge_sink_filter_, gauges_, sinked_gauges_); + f_size(gauges_.size()); + for (auto& gauge : gauges_) { + f_stat(*gauge); + } } void AllocatorImpl::forEachSinkedTextReadout(std::function f_size, std::function f_stat) { Thread::LockGuard lock(mutex_); - forEachSinkedStat(f_size, f_stat, text_readout_sink_filter_, text_readouts_, - sinked_text_readouts_); -} - -void AllocatorImpl::setCounterSinkFilter( - std::function counter_sink_filter) { - ASSERT(counter_sink_filter_ == nullptr); - counter_sink_filter_ = counter_sink_filter; -} - -void AllocatorImpl::setGaugeSinkFilter(std::function gauge_sink_filter) { - ASSERT(gauge_sink_filter_ == nullptr); - gauge_sink_filter_ = gauge_sink_filter; -} - -void AllocatorImpl::setTextReadoutSinkFilter( - std::function text_readout_sink_filter) { - ASSERT(text_readout_sink_filter_ == nullptr); - text_readout_sink_filter_ = text_readout_sink_filter; + f_size(text_readouts_.size()); + for (auto& text_readout : text_readouts_) { + f_stat(*text_readout); + } } } // namespace Stats diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h index a8393444e12bd..b85f8c96e3094 100644 --- a/source/common/stats/allocator_impl.h +++ b/source/common/stats/allocator_impl.h @@ -42,12 +42,6 @@ class AllocatorImpl : public Allocator { void forEachSinkedTextReadout(std::function, std::function) override; - void setCounterSinkFilter(std::function filter) override; - - void setGaugeSinkFilter(std::function filter) override; - - void setTextReadoutSinkFilter(std::function filter) override; - #ifndef ENVOY_CONFIG_COVERAGE void debugPrint(); #endif @@ -73,7 +67,7 @@ class AllocatorImpl : public Allocator { friend class TextReadoutImpl; friend class NotifyingAllocatorImpl; - // We don't need to check StatName to compare sinked stats. + // We don't need to check StatName to compare flushed stats. template using SinkedStatsSet = absl::flat_hash_set>; @@ -81,26 +75,10 @@ class AllocatorImpl : public Allocator { void removeGaugeFromSetLockHeld(Gauge* gauge) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); void removeTextReadoutFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - // Helper method to iterate over sinked stats based on whether or not a stats sink function was - // provided. - template - void forEachSinkedStat(std::function& f_size, - std::function& f_stat, - std::function& sink_filter, - StatSet& stats, SinkedStatsSet& sinked_stats); - StatSet counters_ ABSL_GUARDED_BY(mutex_); StatSet gauges_ ABSL_GUARDED_BY(mutex_); StatSet text_readouts_ ABSL_GUARDED_BY(mutex_); - SinkedStatsSet sinked_counters_ ABSL_GUARDED_BY(mutex_); - SinkedStatsSet sinked_gauges_ ABSL_GUARDED_BY(mutex_); - SinkedStatsSet sinked_text_readouts_ ABSL_GUARDED_BY(mutex_); - - std::function counter_sink_filter_; - std::function gauge_sink_filter_; - std::function text_readout_sink_filter_; - SymbolTable& symbol_table_; // A mutex is needed here to protect both the stats_ object from both @@ -112,27 +90,5 @@ class AllocatorImpl : public Allocator { Thread::ThreadSynchronizer sync_; }; -template -void AllocatorImpl::forEachSinkedStat(std::function& f_size, - std::function& f_stat, - std::function& sink_filter, - StatSet& stats, - SinkedStatsSet& sinked_stats) { - ASSERT(!mutex_.tryLock()); - // Check if a sink filter method was provided. - if (sink_filter != nullptr) { - f_size(sinked_stats.size()); - for (auto& stat : sinked_stats) { - f_stat(*stat); - } - } else { - // Iterate over all stats if a filter method was not provided. - f_size(stats.size()); - for (auto& stat : stats) { - f_stat(*stat); - } - } -} - } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/isolated_store_impl.h b/source/common/stats/isolated_store_impl.h index d5c4c338bc9c7..b8ab5d5059de4 100644 --- a/source/common/stats/isolated_store_impl.h +++ b/source/common/stats/isolated_store_impl.h @@ -46,9 +46,6 @@ template class IsolatedStatsCache { RefcountPtr new_stat = counter_alloc_(name); stats_.emplace(new_stat->statName(), new_stat); - if (sink_filter_ && sink_filter_(*new_stat)) { - sinked_stats_.insert(new_stat.get()); - } return *new_stat; } @@ -60,9 +57,6 @@ template class IsolatedStatsCache { RefcountPtr new_stat = gauge_alloc_(name, import_mode); stats_.emplace(new_stat->statName(), new_stat); - if (sink_filter_ && sink_filter_(*new_stat)) { - sinked_stats_.insert(new_stat.get()); - } return *new_stat; } @@ -85,9 +79,6 @@ template class IsolatedStatsCache { RefcountPtr new_stat = text_readout_alloc_(name, type); stats_.emplace(new_stat->statName(), new_stat); - if (sink_filter_ && sink_filter_(*new_stat)) { - sinked_stats_.insert(new_stat.get()); - } return *new_stat; } @@ -112,21 +103,12 @@ template class IsolatedStatsCache { void forEachSinkedStat(std::function f_size, std::function f_stat) { - if (sink_filter_ != nullptr) { - f_size(sinked_stats_.size()); - for (auto& stat : sinked_stats_) { - f_stat(*stat); - } - } else { - f_size(stats_.size()); - for (auto const& stat : stats_) { - f_stat(*stat.second); - } + f_size(stats_.size()); + for (auto const& stat : stats_) { + f_stat(*stat.second); } } - void setSinkFilter(std::function filter) { sink_filter_ = filter; } - private: friend class IsolatedStoreImpl; @@ -139,12 +121,10 @@ template class IsolatedStatsCache { } StatNameHashMap> stats_; - absl::flat_hash_set> sinked_stats_; CounterAllocator counter_alloc_; GaugeAllocator gauge_alloc_; HistogramAllocator histogram_alloc_; TextReadoutAllocator text_readout_alloc_; - std::function sink_filter_; }; class IsolatedStoreImpl : public StoreImpl { @@ -249,18 +229,6 @@ class IsolatedStoreImpl : public StoreImpl { text_readouts_.forEachSinkedStat(f_size, f_stat); } - void setCounterSinkFilter(std::function filter) override { - counters_.setSinkFilter(filter); - } - - void setGaugeSinkFilter(std::function filter) override { - gauges_.setSinkFilter(filter); - } - - void setTextReadoutSinkFilter(std::function filter) override { - text_readouts_.setSinkFilter(filter); - } - private: IsolatedStoreImpl(std::unique_ptr&& symbol_table); diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc index 1e90d937b5ee8..1690383ef5953 100644 --- a/source/common/stats/thread_local_store.cc +++ b/source/common/stats/thread_local_store.cc @@ -992,18 +992,5 @@ void ThreadLocalStoreImpl::forEachSinkedTextReadout( alloc_.forEachSinkedTextReadout(f_size, f_stat); } -void ThreadLocalStoreImpl::setCounterSinkFilter(std::function filter) { - alloc_.setCounterSinkFilter(filter); -} - -void ThreadLocalStoreImpl::setGaugeSinkFilter(std::function filter) { - alloc_.setGaugeSinkFilter(filter); -} - -void ThreadLocalStoreImpl::setTextReadoutSinkFilter( - std::function filter) { - alloc_.setTextReadoutSinkFilter(filter); -} - } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h index 18b983ac834dd..ad16081b63c97 100644 --- a/source/common/stats/thread_local_store.h +++ b/source/common/stats/thread_local_store.h @@ -253,12 +253,6 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo void forEachSinkedTextReadout(std::function f_size, std::function f_stat) override; - void setCounterSinkFilter(std::function) override; - - void setGaugeSinkFilter(std::function) override; - - void setTextReadoutSinkFilter(std::function) override; - // Stats::StoreRoot void addSink(Sink& sink) override { timer_sinks_.push_back(sink); } void setTagProducer(TagProducerPtr&& tag_producer) override { diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index 0991f6d94ab23..83fb85d68ea8d 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -1,7 +1,5 @@ #include -#include "envoy/stats/stats.h" - #include "source/common/stats/allocator_impl.h" #include "test/test_common/logging.h" @@ -127,142 +125,6 @@ TEST_F(AllocatorImplTest, RefCountDecAllocRaceSynchronized) { EXPECT_FALSE(alloc_.isMutexLockedForTest()); } -TEST_F(AllocatorImplTest, ForEachSinkedCounter) { - - StatNameHashSet sinked_stat_names; - std::vector sinked_counters; - std::vector unsinked_counters; - - alloc_.setCounterSinkFilter([&sinked_stat_names](const Stats::Counter& counter) { - if (sinked_stat_names.count(counter.statName()) == 1) { - return true; - } - return false; - }); - - size_t n_stats = 11; - - for (size_t idx = 0; idx < n_stats; ++idx) { - auto stat_name = makeStat(absl::StrCat("counter.", idx)); - // sink every 3rd stat - if ((idx + 1) % 3 == 0) { - sinked_stat_names.insert(stat_name); - sinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); - } else { - unsinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); - } - } - - size_t n_sinked_counters = 0; - size_t n_iterations = 0; - alloc_.forEachSinkedCounter([&n_sinked_counters](std::size_t size) { n_sinked_counters = size; }, - [&n_iterations, &sinked_stat_names](Stats::Counter& counter) { - EXPECT_EQ(sinked_stat_names.count(counter.statName()), 1); - ++n_iterations; - }); - EXPECT_EQ(n_sinked_counters, 3); - - // Erase all sinked stats. - sinked_counters.clear(); - n_iterations = 0; - alloc_.forEachSinkedCounter([&n_sinked_counters](std::size_t size) { n_sinked_counters = size; }, - [&n_iterations](Stats::Counter&) { ++n_iterations; }); - EXPECT_EQ(n_sinked_counters, 0); - EXPECT_EQ(n_iterations, 0); -} - -TEST_F(AllocatorImplTest, ForEachSinkedGauge) { - - StatNameHashSet sinked_stat_names; - std::vector sinked_gauges; - std::vector unsinked_gauges; - - alloc_.setGaugeSinkFilter([&sinked_stat_names](const Stats::Gauge& gauge) { - if (sinked_stat_names.count(gauge.statName()) == 1) { - return true; - } - return false; - }); - - size_t n_stats = 11; - - for (size_t idx = 0; idx < n_stats; ++idx) { - auto stat_name = makeStat(absl::StrCat("gauge.", idx)); - // sink every 5th stat - if ((idx + 1) % 5 == 0) { - sinked_stat_names.insert(stat_name); - sinked_gauges.emplace_back( - alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); - } else { - unsinked_gauges.emplace_back( - alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); - } - } - - size_t n_sinked_gauges = 0; - size_t n_iterations = 0; - alloc_.forEachSinkedGauge([&n_sinked_gauges](std::size_t size) { n_sinked_gauges = size; }, - [&n_iterations, &sinked_stat_names](Stats::Gauge& gauge) { - EXPECT_EQ(sinked_stat_names.count(gauge.statName()), 1); - ++n_iterations; - }); - EXPECT_EQ(n_sinked_gauges, 2); - - // Erase all sinked stats. - sinked_gauges.clear(); - n_iterations = 0; - alloc_.forEachSinkedGauge([&n_sinked_gauges](std::size_t size) { n_sinked_gauges = size; }, - [&n_iterations](Stats::Gauge&) { ++n_iterations; }); - EXPECT_EQ(n_sinked_gauges, 0); - EXPECT_EQ(n_iterations, 0); -} - -TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) { - - StatNameHashSet sinked_stat_names; - std::vector sinked_text_readouts; - std::vector unsinked_text_readouts; - - alloc_.setTextReadoutSinkFilter([&sinked_stat_names](const Stats::TextReadout& text_readout) { - if (sinked_stat_names.count(text_readout.statName()) == 1) { - return true; - } - return false; - }); - - size_t n_stats = 11; - - for (size_t idx = 0; idx < n_stats; ++idx) { - auto stat_name = makeStat(absl::StrCat("text_readout.", idx)); - // sink every 2nd stat - if ((idx + 1) % 2 == 0) { - sinked_stat_names.insert(stat_name); - sinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); - } else { - unsinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); - } - } - - size_t n_sinked_text_readouts = 0; - size_t n_iterations = 0; - alloc_.forEachSinkedTextReadout( - [&n_sinked_text_readouts](std::size_t size) { n_sinked_text_readouts = size; }, - [&n_iterations, &sinked_stat_names](Stats::TextReadout& text_readout) { - EXPECT_EQ(sinked_stat_names.count(text_readout.statName()), 1); - ++n_iterations; - }); - EXPECT_EQ(n_sinked_text_readouts, 5); - - // Erase all sinked stats. - sinked_text_readouts.clear(); - n_iterations = 0; - alloc_.forEachSinkedTextReadout( - [&n_sinked_text_readouts](std::size_t size) { n_sinked_text_readouts = size; }, - [&n_iterations](Stats::TextReadout&) { ++n_iterations; }); - EXPECT_EQ(n_sinked_text_readouts, 0); - EXPECT_EQ(n_iterations, 0); -} - } // namespace } // namespace Stats } // namespace Envoy diff --git a/test/integration/server.h b/test/integration/server.h index cfd18f017a4fa..21415e9997e9d 100644 --- a/test/integration/server.h +++ b/test/integration/server.h @@ -299,18 +299,6 @@ class TestIsolatedStoreImpl : public StoreRoot { store_.forEachSinkedTextReadout(f_size, f_stat); } - void setCounterSinkFilter(std::function filter) override { - store_.setCounterSinkFilter(filter); - } - - void setGaugeSinkFilter(std::function filter) override { - store_.setGaugeSinkFilter(filter); - } - - void setTextReadoutSinkFilter(std::function filter) override { - store_.setTextReadoutSinkFilter(filter); - } - Counter& counterFromString(const std::string& name) override { Thread::LockGuard lock(lock_); return store_.counterFromString(name); diff --git a/test/mocks/stats/mocks.h b/test/mocks/stats/mocks.h index 5e61f39709cd1..85894a7144b4c 100644 --- a/test/mocks/stats/mocks.h +++ b/test/mocks/stats/mocks.h @@ -292,9 +292,6 @@ class MockStore : public TestUtil::TestStore { (std::function, std::function)); MOCK_METHOD(void, forEachSinkedTextReadout, (std::function, std::function)); - MOCK_METHOD(void, setCounterSinkFilter, (std::function)); - MOCK_METHOD(void, setGaugeSinkFilter, (std::function)); - MOCK_METHOD(void, setTextReadoutSinkFilter, (std::function)); MOCK_METHOD(CounterOptConstRef, findCounter, (StatName), (const)); MOCK_METHOD(GaugeOptConstRef, findGauge, (StatName), (const)); From d556414a281bf55d9d026be4782eef4ecdeb62c5 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Mon, 30 Aug 2021 17:31:07 +0000 Subject: [PATCH 04/12] Revert added test for phase 2 Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 2 +- .../server_stats_flush_benchmark_test.cc | 22 +++----- test/server/server_test.cc | 51 ------------------- 3 files changed, 8 insertions(+), 67 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index d3fab4cd6024f..33ac4531e0039 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -62,7 +62,7 @@ class Allocator { * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. * @param f_size functor that is provided the number of all sinked stats. Note this is called - * only once. + * only once, prior to any calls to f_stat. * @param f_stat functor that is provided one sinked stat at a time. */ virtual void forEachSinkedCounter(std::function f_size, diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index 879a5beef7fec..9d6bac2b6a741 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -18,23 +18,23 @@ namespace Envoy { class StatsSinkFlushSpeedTest { public: - StatsSinkFlushSpeedTest(uint64_t n_counters, uint64_t n_gauges, uint64_t n_text_readouts) + StatsSinkFlushSpeedTest(size_t const num_stats) : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { sinks_.emplace_back(new testing::NiceMock()); // Create counters - for (uint64_t idx = 0; idx < n_counters; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("counter.", idx)); stats_store_.counterFromStatName(stat_name).inc(); } // Create gauges - for (uint64_t idx = 0; idx < n_gauges; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("gauge.", idx)); stats_store_.gaugeFromStatName(stat_name, Stats::Gauge::ImportMode::NeverImport).set(idx); } // Create text readouts - for (uint64_t idx = 0; idx < n_text_readouts; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("text_readout.", idx)); stats_store_.textReadoutFromStatName(stat_name).set(absl::StrCat("text_readout.", idx)); } @@ -56,18 +56,10 @@ class StatsSinkFlushSpeedTest { Event::SimulatedTimeSystem time_system_; }; -static void bmLarge(benchmark::State& state) { - uint64_t n_counters = 1000000, n_gauges = 1000000, n_text_readouts = 1000000; - StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); +static void bmFlushToSinks(benchmark::State& state) { + StatsSinkFlushSpeedTest speed_test(state.range(0)); speed_test.test(state); } -BENCHMARK(bmLarge)->Unit(::benchmark::kMillisecond); - -static void bmSmall(benchmark::State& state) { - uint64_t n_counters = 10000, n_gauges = 10000, n_text_readouts = 10000; - StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); - speed_test.test(state); -} -BENCHMARK(bmSmall)->Unit(::benchmark::kMillisecond); +BENCHMARK(bmFlushToSinks)->Unit(::benchmark::kMillisecond)->RangeMultiplier(10)->Range(10, 1000000); } // namespace Envoy diff --git a/test/server/server_test.cc b/test/server/server_test.cc index b57247e67609b..8e6acff4fe85d 100644 --- a/test/server/server_test.cc +++ b/test/server/server_test.cc @@ -101,57 +101,6 @@ TEST(ServerInstanceUtil, flushHelper) { InstanceUtil::flushMetricsToSinks(sinks, mock_store, time_system); } -TEST(ServerInstanceUtil, sinkFilter) { - InSequence s; - - Stats::TestUtil::TestStore store; - Event::SimulatedTimeSystem time_system; - // Set the sink filters - store.setCounterSinkFilter( - [](const Stats::Counter& counter) { return counter.name().compare("hello") == 0; }); - store.setGaugeSinkFilter( - [](const Stats::Gauge& gauge) { return gauge.name().compare("day") == 0; }); - store.setTextReadoutSinkFilter([](const Stats::TextReadout& text_readout) { - return text_readout.name().compare("text") == 0; - }); - - Stats::Counter& c = store.counter("hello"); - c.inc(); - c.inc(); - store.counter("world").inc(); - - store.gauge("sunny", Stats::Gauge::ImportMode::Accumulate).set(9); - store.gauge("day", Stats::Gauge::ImportMode::Accumulate).set(5); - - store.textReadout("Text").set("don't sink"); - store.textReadout("text").set("is important"); - - std::list sinks; - InstanceUtil::flushMetricsToSinks(sinks, store, time_system); - // Make sure that counters have been latched even if there are no sinks. - EXPECT_EQ(2UL, c.value()); - EXPECT_EQ(0, c.latch()); - - Stats::MockSink* sink = new StrictMock(); - sinks.emplace_back(sink); - EXPECT_CALL(*sink, flush(_)).WillOnce(Invoke([](Stats::MetricSnapshot& snapshot) { - ASSERT_EQ(snapshot.counters().size(), 1); - EXPECT_EQ(snapshot.counters()[0].counter_.get().name(), "hello"); - EXPECT_EQ(snapshot.counters()[0].delta_, 2); - - ASSERT_EQ(snapshot.gauges().size(), 1); - EXPECT_EQ(snapshot.gauges()[0].get().name(), "day"); - EXPECT_EQ(snapshot.gauges()[0].get().value(), 5); - - ASSERT_EQ(snapshot.textReadouts().size(), 1); - EXPECT_EQ(snapshot.textReadouts()[0].get().name(), "text"); - EXPECT_EQ(snapshot.textReadouts()[0].get().value(), "is important"); - })); - c.inc(); - c.inc(); - InstanceUtil::flushMetricsToSinks(sinks, store, time_system); -} - class RunHelperTest : public testing::Test { public: RunHelperTest() { From f91bb29bdd5fe75fe8b1bf6a3434fa408b655d77 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Mon, 30 Aug 2021 17:31:07 +0000 Subject: [PATCH 05/12] Revert added test for phase 2 Return early in benchmark for unit tests to prevent timeout. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 2 +- .../server_stats_flush_benchmark_test.cc | 30 +++++------ test/server/server_test.cc | 51 ------------------- 3 files changed, 16 insertions(+), 67 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index d3fab4cd6024f..33ac4531e0039 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -62,7 +62,7 @@ class Allocator { * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. * @param f_size functor that is provided the number of all sinked stats. Note this is called - * only once. + * only once, prior to any calls to f_stat. * @param f_stat functor that is provided one sinked stat at a time. */ virtual void forEachSinkedCounter(std::function f_size, diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index 879a5beef7fec..edc4b567456ef 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -7,40 +7,42 @@ #include "source/common/stats/thread_local_store.h" #include "source/server/server.h" +#include "test/benchmark/main.h" #include "test/mocks/stats/mocks.h" #include "test/test_common/simulated_time_system.h" #include "absl/strings/str_cat.h" #include "benchmark/benchmark.h" #include "gmock/gmock.h" +#include "gtest/gtest.h" namespace Envoy { class StatsSinkFlushSpeedTest { public: - StatsSinkFlushSpeedTest(uint64_t n_counters, uint64_t n_gauges, uint64_t n_text_readouts) + StatsSinkFlushSpeedTest(size_t const num_stats) : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { sinks_.emplace_back(new testing::NiceMock()); // Create counters - for (uint64_t idx = 0; idx < n_counters; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("counter.", idx)); stats_store_.counterFromStatName(stat_name).inc(); } // Create gauges - for (uint64_t idx = 0; idx < n_gauges; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("gauge.", idx)); stats_store_.gaugeFromStatName(stat_name, Stats::Gauge::ImportMode::NeverImport).set(idx); } // Create text readouts - for (uint64_t idx = 0; idx < n_text_readouts; ++idx) { + for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("text_readout.", idx)); stats_store_.textReadoutFromStatName(stat_name).set(absl::StrCat("text_readout.", idx)); } } - void test(benchmark::State& state) { + void test(::benchmark::State& state) { for (auto _ : state) { UNREFERENCED_PARAMETER(state); Server::InstanceUtil::flushMetricsToSinks(sinks_, stats_store_, time_system_); @@ -56,18 +58,16 @@ class StatsSinkFlushSpeedTest { Event::SimulatedTimeSystem time_system_; }; -static void bmLarge(benchmark::State& state) { - uint64_t n_counters = 1000000, n_gauges = 1000000, n_text_readouts = 1000000; - StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); - speed_test.test(state); -} -BENCHMARK(bmLarge)->Unit(::benchmark::kMillisecond); +static void bmFlushToSinks(::benchmark::State& state) { + // Skip expensive benchmarks for unit tests. + if (benchmark::skipExpensiveBenchmarks() && state.range(0) > 100) { + state.SkipWithError("Skipping expensive benchmark"); + return; + } -static void bmSmall(benchmark::State& state) { - uint64_t n_counters = 10000, n_gauges = 10000, n_text_readouts = 10000; - StatsSinkFlushSpeedTest speed_test(n_counters, n_gauges, n_text_readouts); + StatsSinkFlushSpeedTest speed_test(state.range(0)); speed_test.test(state); } -BENCHMARK(bmSmall)->Unit(::benchmark::kMillisecond); +BENCHMARK(bmFlushToSinks)->Unit(::benchmark::kMillisecond)->RangeMultiplier(10)->Range(10, 1000000); } // namespace Envoy diff --git a/test/server/server_test.cc b/test/server/server_test.cc index b57247e67609b..8e6acff4fe85d 100644 --- a/test/server/server_test.cc +++ b/test/server/server_test.cc @@ -101,57 +101,6 @@ TEST(ServerInstanceUtil, flushHelper) { InstanceUtil::flushMetricsToSinks(sinks, mock_store, time_system); } -TEST(ServerInstanceUtil, sinkFilter) { - InSequence s; - - Stats::TestUtil::TestStore store; - Event::SimulatedTimeSystem time_system; - // Set the sink filters - store.setCounterSinkFilter( - [](const Stats::Counter& counter) { return counter.name().compare("hello") == 0; }); - store.setGaugeSinkFilter( - [](const Stats::Gauge& gauge) { return gauge.name().compare("day") == 0; }); - store.setTextReadoutSinkFilter([](const Stats::TextReadout& text_readout) { - return text_readout.name().compare("text") == 0; - }); - - Stats::Counter& c = store.counter("hello"); - c.inc(); - c.inc(); - store.counter("world").inc(); - - store.gauge("sunny", Stats::Gauge::ImportMode::Accumulate).set(9); - store.gauge("day", Stats::Gauge::ImportMode::Accumulate).set(5); - - store.textReadout("Text").set("don't sink"); - store.textReadout("text").set("is important"); - - std::list sinks; - InstanceUtil::flushMetricsToSinks(sinks, store, time_system); - // Make sure that counters have been latched even if there are no sinks. - EXPECT_EQ(2UL, c.value()); - EXPECT_EQ(0, c.latch()); - - Stats::MockSink* sink = new StrictMock(); - sinks.emplace_back(sink); - EXPECT_CALL(*sink, flush(_)).WillOnce(Invoke([](Stats::MetricSnapshot& snapshot) { - ASSERT_EQ(snapshot.counters().size(), 1); - EXPECT_EQ(snapshot.counters()[0].counter_.get().name(), "hello"); - EXPECT_EQ(snapshot.counters()[0].delta_, 2); - - ASSERT_EQ(snapshot.gauges().size(), 1); - EXPECT_EQ(snapshot.gauges()[0].get().name(), "day"); - EXPECT_EQ(snapshot.gauges()[0].get().value(), 5); - - ASSERT_EQ(snapshot.textReadouts().size(), 1); - EXPECT_EQ(snapshot.textReadouts()[0].get().name(), "text"); - EXPECT_EQ(snapshot.textReadouts()[0].get().value(), "is important"); - })); - c.inc(); - c.inc(); - InstanceUtil::flushMetricsToSinks(sinks, store, time_system); -} - class RunHelperTest : public testing::Test { public: RunHelperTest() { From c847d1c4fd052d1bcb3eb26de3a8303c82ba7406 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Tue, 31 Aug 2021 13:01:33 +0000 Subject: [PATCH 06/12] Add missing header Signed-off-by: Pradeep Rao --- test/server/server_stats_flush_benchmark_test.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index edc4b567456ef..34e6c57171517 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -10,6 +10,7 @@ #include "test/benchmark/main.h" #include "test/mocks/stats/mocks.h" #include "test/test_common/simulated_time_system.h" +#include "test/test_common/utility.h" #include "absl/strings/str_cat.h" #include "benchmark/benchmark.h" From 291a2ffe27a70143a8384ecc377a891b41e2ca51 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Wed, 1 Sep 2021 02:11:41 +0000 Subject: [PATCH 07/12] Keep rejected counters, gauges and text readouts in the allocator, instead of the store. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 20 +++-- envoy/stats/store.h | 12 +-- source/common/common/interval_value.h | 9 +- source/common/local_reply/local_reply.cc | 11 +-- source/common/stats/allocator_impl.cc | 64 ++++++++++++-- source/common/stats/allocator_impl.h | 35 +++++--- source/common/stats/isolated_store_impl.h | 22 ++--- source/common/stats/thread_local_store.cc | 87 +++++++++---------- source/common/stats/thread_local_store.h | 17 ++-- source/server/server.cc | 20 ++--- test/common/network/apple_dns_impl_test.cc | 2 +- test/common/stats/thread_local_store_test.cc | 3 +- test/mocks/stats/mocks.h | 13 +-- .../server_stats_flush_benchmark_test.cc | 7 +- 14 files changed, 196 insertions(+), 126 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index 33ac4531e0039..3f5ad4112c5ad 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -58,6 +58,14 @@ class Allocator { virtual const SymbolTable& constSymbolTable() const PURE; virtual SymbolTable& symbolTable() PURE; + /** + * Mark rejected stats as deleted by moving them to a different vector, so they don't show up + * when iterating over stats, but prevent crashes when trying to accesses references to them. + */ + virtual void markCounterForDeletion(StatName name) PURE; + virtual void markGaugeForDeletion(StatName name) PURE; + virtual void markTextReadoutForDeletion(StatName name) PURE; + /** * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold * on to a mutex that will deadlock if the passed in functors try to create or delete a stat. @@ -65,12 +73,12 @@ class Allocator { * only once, prior to any calls to f_stat. * @param f_stat functor that is provided one sinked stat at a time. */ - virtual void forEachSinkedCounter(std::function f_size, - std::function f_stat) PURE; - virtual void forEachSinkedGauge(std::function f_size, - std::function f_stat) PURE; - virtual void forEachSinkedTextReadout(std::function f_size, - std::function f_stat) PURE; + virtual void forEachCounter(std::function f_size, + std::function f_stat) const PURE; + virtual void forEachGauge(std::function f_size, + std::function f_stat) const PURE; + virtual void forEachTextReadout(std::function f_size, + std::function f_stat) const PURE; // TODO(jmarantz): create a parallel mechanism to instantiate histograms. At // the moment, histograms don't fit the same pattern of counters and gauges diff --git a/envoy/stats/store.h b/envoy/stats/store.h index 82cb9e20ba5c3..a682fb0cd3d5f 100644 --- a/envoy/stats/store.h +++ b/envoy/stats/store.h @@ -56,14 +56,14 @@ class Store : public Scope { * @param f_size functor that is provided the number of all sinked stats. * @param f_stat functor that is provided one sinked stat at a time. */ - virtual void forEachSinkedCounter(std::function f_size, - std::function f_stat) PURE; + virtual void forEachCounter(std::function f_size, + std::function f_stat) const PURE; - virtual void forEachSinkedGauge(std::function f_size, - std::function f_stat) PURE; + virtual void forEachGauge(std::function f_size, + std::function f_stat) const PURE; - virtual void forEachSinkedTextReadout(std::function f_size, - std::function f_stat) PURE; + virtual void forEachTextReadout(std::function f_size, + std::function f_stat) const PURE; }; using StorePtr = std::unique_ptr; diff --git a/source/common/common/interval_value.h b/source/common/common/interval_value.h index e001a8a13a324..3a058eaae4b54 100644 --- a/source/common/common/interval_value.h +++ b/source/common/common/interval_value.h @@ -25,10 +25,11 @@ template class ClosedIntervalValue { // Returns a value that is as far from max as the original value is from min. // This guarantees that max().invert() == min() and min().invert() == max(). ClosedIntervalValue invert() const { - return ClosedIntervalValue(value_ == Interval::max_value ? Interval::min_value - : value_ == Interval::min_value - ? Interval::max_value - : Interval::max_value - (value_ - Interval::min_value)); + return ClosedIntervalValue(value_ == Interval::max_value + ? Interval::min_value + : value_ == Interval::min_value + ? Interval::max_value + : Interval::max_value - (value_ - Interval::min_value)); } // Comparisons are performed using the same operators on the underlying value diff --git a/source/common/local_reply/local_reply.cc b/source/common/local_reply/local_reply.cc index 0f8b2db588677..365f9ab36abd0 100644 --- a/source/common/local_reply/local_reply.cc +++ b/source/common/local_reply/local_reply.cc @@ -26,11 +26,12 @@ class BodyFormatter { Server::Configuration::CommonFactoryContext& context) : formatter_(Formatter::SubstitutionFormatStringUtils::fromProtoConfig(config, context)), content_type_( - !config.content_type().empty() ? config.content_type() - : config.format_case() == - envoy::config::core::v3::SubstitutionFormatString::FormatCase::kJsonFormat - ? Http::Headers::get().ContentTypeValues.Json - : Http::Headers::get().ContentTypeValues.Text) {} + !config.content_type().empty() + ? config.content_type() + : config.format_case() == + envoy::config::core::v3::SubstitutionFormatString::FormatCase::kJsonFormat + ? Http::Headers::get().ContentTypeValues.Json + : Http::Headers::get().ContentTypeValues.Text) {} void format(const Http::RequestHeaderMap& request_headers, const Http::ResponseHeaderMap& response_headers, diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index 46c97797ee451..ff093788c8ca3 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -17,6 +17,19 @@ #include "absl/container/flat_hash_set.h" +namespace { +// Helper function to check if vector of StatSharedPtrs contains a StatPtr +// Note this is only meant to be used in asserts as this check is expensive. +template +bool hasStat(const std::vector>& vec, + StatType* const stat_ptr) { + return std::find_if(vec.begin(), vec.end(), + [stat_ptr](const Envoy::Stats::RefcountPtr& shared_ptr) { + return (shared_ptr.get() == stat_ptr); + }) != vec.end(); +} +} // namespace + namespace Envoy { namespace Stats { @@ -123,7 +136,7 @@ class CounterImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.counters_.erase(statName()); - ASSERT(count == 1); + ASSERT(count == 1 || hasStat(alloc_.deleted_counters_, this)); } // Stats::Counter @@ -167,7 +180,7 @@ class GaugeImpl : public StatsSharedImpl { void removeFromSetLockHeld() override ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) { const size_t count = alloc_.gauges_.erase(statName()); - ASSERT(count == 1); + ASSERT(count == 1 || hasStat(alloc_.deleted_gauges_, this)); } // Stats::Gauge @@ -239,7 +252,7 @@ class TextReadoutImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.text_readouts_.erase(statName()); - ASSERT(count == 1); + ASSERT(count == 1 || hasStat(alloc_.deleted_text_readouts_, this)); } // Stats::TextReadout @@ -316,8 +329,8 @@ Counter* AllocatorImpl::makeCounterInternal(StatName name, StatName tag_extracte return new CounterImpl(name, *this, tag_extracted_name, stat_name_tags); } -void AllocatorImpl::forEachSinkedCounter(std::function f_size, - std::function f_stat) { +void AllocatorImpl::forEachCounter(std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(mutex_); f_size(counters_.size()); for (auto& counter : counters_) { @@ -325,8 +338,8 @@ void AllocatorImpl::forEachSinkedCounter(std::function f_size } } -void AllocatorImpl::forEachSinkedGauge(std::function f_size, - std::function f_stat) { +void AllocatorImpl::forEachGauge(std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(mutex_); f_size(gauges_.size()); for (auto& gauge : gauges_) { @@ -334,8 +347,8 @@ void AllocatorImpl::forEachSinkedGauge(std::function f_size, } } -void AllocatorImpl::forEachSinkedTextReadout(std::function f_size, - std::function f_stat) { +void AllocatorImpl::forEachTextReadout(std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(mutex_); f_size(text_readouts_.size()); for (auto& text_readout : text_readouts_) { @@ -343,5 +356,38 @@ void AllocatorImpl::forEachSinkedTextReadout(std::function f_ } } +void AllocatorImpl::markCounterForDeletion(StatName name) { + Thread::LockGuard lock(mutex_); + auto iter = counters_.find(name); + if (iter == counters_.end()) { + return; + } + ASSERT(!hasStat(deleted_counters_, *iter)); + deleted_counters_.emplace_back(*iter); + counters_.erase(iter); +} + +void AllocatorImpl::markGaugeForDeletion(StatName name) { + Thread::LockGuard lock(mutex_); + auto iter = gauges_.find(name); + if (iter == gauges_.end()) { + return; + } + ASSERT(!hasStat(deleted_gauges_, *iter)); + deleted_gauges_.emplace_back(*iter); + gauges_.erase(iter); +} + +void AllocatorImpl::markTextReadoutForDeletion(StatName name) { + Thread::LockGuard lock(mutex_); + auto iter = text_readouts_.find(name); + if (iter == text_readouts_.end()) { + return; + } + ASSERT(!hasStat(deleted_text_readouts_, *iter)); + deleted_text_readouts_.emplace_back(*iter); + text_readouts_.erase(iter); +} + } // namespace Stats } // namespace Envoy diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h index b85f8c96e3094..c0a62eac16770 100644 --- a/source/common/stats/allocator_impl.h +++ b/source/common/stats/allocator_impl.h @@ -11,6 +11,7 @@ #include "absl/container/flat_hash_set.h" #include "absl/strings/string_view.h" +#include "metric_impl.h" namespace Envoy { namespace Stats { @@ -33,14 +34,14 @@ class AllocatorImpl : public Allocator { SymbolTable& symbolTable() override { return symbol_table_; } const SymbolTable& constSymbolTable() const override { return symbol_table_; } - void forEachSinkedCounter(std::function, - std::function) override; + void forEachCounter(std::function, + std::function) const override; - void forEachSinkedGauge(std::function, - std::function) override; + void forEachGauge(std::function, + std::function) const override; - void forEachSinkedTextReadout(std::function, - std::function) override; + void forEachTextReadout(std::function, + std::function) const override; #ifndef ENVOY_CONFIG_COVERAGE void debugPrint(); @@ -56,6 +57,10 @@ class AllocatorImpl : public Allocator { */ bool isMutexLockedForTest(); + void markCounterForDeletion(StatName name) override; + void markGaugeForDeletion(StatName name) override; + void markTextReadoutForDeletion(StatName name) override; + protected: virtual Counter* makeCounterInternal(StatName name, StatName tag_extracted_name, const StatNameTagVector& stat_name_tags); @@ -71,21 +76,29 @@ class AllocatorImpl : public Allocator { template using SinkedStatsSet = absl::flat_hash_set>; - void removeCounterFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - void removeGaugeFromSetLockHeld(Gauge* gauge) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - void removeTextReadoutFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - StatSet counters_ ABSL_GUARDED_BY(mutex_); StatSet gauges_ ABSL_GUARDED_BY(mutex_); StatSet text_readouts_ ABSL_GUARDED_BY(mutex_); + // Retain storage for deleted stats; these are no longer in maps because + // the matcher-pattern was established after they were created. Since the + // stats are held by reference in code that expects them to be there, we + // can't actually delete the stats. + // + // It seems like it would be better to have each client that expects a stat + // to exist to hold it as (e.g.) a CounterSharedPtr rather than a Counter& + // but that would be fairly complex to change. + std::vector deleted_counters_ ABSL_GUARDED_BY(mutex_); + std::vector deleted_gauges_ ABSL_GUARDED_BY(mutex_); + std::vector deleted_text_readouts_ ABSL_GUARDED_BY(mutex_); + SymbolTable& symbol_table_; // A mutex is needed here to protect both the stats_ object from both // alloc() and free() operations. Although alloc() operations are called under existing locking, // free() operations are made from the destructors of the individual stat objects, which are not // protected by locks. - Thread::MutexBasicLockable mutex_; + mutable Thread::MutexBasicLockable mutex_; Thread::ThreadSynchronizer sync_; }; diff --git a/source/common/stats/isolated_store_impl.h b/source/common/stats/isolated_store_impl.h index b8ab5d5059de4..ebff944da7eff 100644 --- a/source/common/stats/isolated_store_impl.h +++ b/source/common/stats/isolated_store_impl.h @@ -101,8 +101,8 @@ template class IsolatedStatsCache { return true; } - void forEachSinkedStat(std::function f_size, - std::function f_stat) { + void forEachStat(std::function f_size, + std::function f_stat) const { f_size(stats_.size()); for (auto const& stat : stats_) { f_stat(*stat.second); @@ -214,19 +214,19 @@ class IsolatedStoreImpl : public StoreImpl { return textReadoutFromStatName(storage.statName()); } - void forEachSinkedCounter(std::function f_size, - std::function f_stat) override { - counters_.forEachSinkedStat(f_size, f_stat); + void forEachCounter(std::function f_size, + std::function f_stat) const override { + counters_.forEachStat(f_size, f_stat); } - void forEachSinkedGauge(std::function f_size, - std::function f_stat) override { - gauges_.forEachSinkedStat(f_size, f_stat); + void forEachGauge(std::function f_size, + std::function f_stat) const override { + gauges_.forEachStat(f_size, f_stat); } - void forEachSinkedTextReadout(std::function f_size, - std::function f_stat) override { - text_readouts_.forEachSinkedStat(f_size, f_stat); + void forEachTextReadout(std::function f_size, + std::function f_stat) const override { + text_readouts_.forEachStat(f_size, f_stat); } private: diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc index 1690383ef5953..0db5c94cb60df 100644 --- a/source/common/stats/thread_local_store.cc +++ b/source/common/stats/thread_local_store.cc @@ -69,10 +69,13 @@ void ThreadLocalStoreImpl::setStatsMatcher(StatsMatcherPtr&& stats_matcher) { Thread::LockGuard lock(lock_); const uint32_t first_histogram_index = deleted_histograms_.size(); for (ScopeImpl* scope : scopes_) { - removeRejectedStats(scope->central_cache_->counters_, deleted_counters_); - removeRejectedStats(scope->central_cache_->gauges_, deleted_gauges_); + removeRejectedStats(scope->central_cache_->counters_, + [this](StatName name) mutable { alloc_.markCounterForDeletion(name); }); + removeRejectedStats(scope->central_cache_->gauges_, + [this](StatName name) mutable { alloc_.markGaugeForDeletion(name); }); removeRejectedStats(scope->central_cache_->histograms_, deleted_histograms_); - removeRejectedStats(scope->central_cache_->text_readouts_, deleted_text_readouts_); + removeRejectedStats(scope->central_cache_->text_readouts_, + [this](StatName name) mutable { alloc_.markTextReadoutForDeletion(name); }); } // Remove any newly rejected histograms from histogram_set_. @@ -101,6 +104,23 @@ void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, StatListClass& } } +template +void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, + std::function f_deletion) { + StatNameVec remove_list; + for (auto& stat : map) { + if (rejects(stat.first)) { + remove_list.push_back(stat.first); + } + } + for (StatName stat_name : remove_list) { + auto iter = map.find(stat_name); + ASSERT(iter != map.end()); + f_deletion(stat_name); + map.erase(iter); + } +} + StatsMatcher::FastResult ThreadLocalStoreImpl::fastRejects(StatName stat_name) const { return stats_matcher_->fastRejects(stat_name); } @@ -113,16 +133,9 @@ bool ThreadLocalStoreImpl::slowRejects(StatsMatcher::FastResult fast_reject_resu std::vector ThreadLocalStoreImpl::counters() const { // Handle de-dup due to overlapping scopes. std::vector ret; - StatNameHashSet names; - Thread::LockGuard lock(lock_); - for (ScopeImpl* scope : scopes_) { - for (auto& counter : scope->central_cache_->counters_) { - if (names.insert(counter.first).second) { - ret.push_back(counter.second); - } - } - } - + forEachCounter( + [&ret](std::size_t size) mutable { ret.reserve(size); }, + [&ret](Counter& counter) mutable { ret.emplace_back(CounterSharedPtr(&counter)); }); return ret; } @@ -141,34 +154,18 @@ ScopePtr ThreadLocalStoreImpl::scopeFromStatName(StatName name) { std::vector ThreadLocalStoreImpl::gauges() const { // Handle de-dup due to overlapping scopes. std::vector ret; - StatNameHashSet names; - Thread::LockGuard lock(lock_); - for (ScopeImpl* scope : scopes_) { - for (auto& gauge_iter : scope->central_cache_->gauges_) { - const GaugeSharedPtr& gauge = gauge_iter.second; - if (gauge->importMode() != Gauge::ImportMode::Uninitialized && - names.insert(gauge_iter.first).second) { - ret.push_back(gauge); - } - } - } - + forEachGauge([&ret](std::size_t size) mutable { ret.reserve(size); }, + [&ret](Gauge& gauge) mutable { ret.emplace_back(GaugeSharedPtr(&gauge)); }); return ret; } std::vector ThreadLocalStoreImpl::textReadouts() const { // Handle de-dup due to overlapping scopes. std::vector ret; - StatNameHashSet names; - Thread::LockGuard lock(lock_); - for (ScopeImpl* scope : scopes_) { - for (auto& text_readout : scope->central_cache_->text_readouts_) { - if (names.insert(text_readout.first).second) { - ret.push_back(text_readout.second); - } - } - } - + forEachTextReadout([&ret](std::size_t size) mutable { ret.reserve(size); }, + [&ret](TextReadout& text_readout) mutable { + ret.emplace_back(TextReadoutSharedPtr(&text_readout)); + }); return ret; } @@ -975,21 +972,23 @@ bool ParentHistogramImpl::usedLockHeld() const { return false; } -void ThreadLocalStoreImpl::forEachSinkedCounter(std::function f_size, - std::function f_stat) { +void ThreadLocalStoreImpl::forEachCounter(std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(lock_); - alloc_.forEachSinkedCounter(f_size, f_stat); + alloc_.forEachCounter(f_size, f_stat); } -void ThreadLocalStoreImpl::forEachSinkedGauge(std::function f_size, - std::function f_stat) { +void ThreadLocalStoreImpl::forEachGauge(std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(lock_); - alloc_.forEachSinkedGauge(f_size, f_stat); + alloc_.forEachGauge(f_size, f_stat); } -void ThreadLocalStoreImpl::forEachSinkedTextReadout( - std::function f_size, std::function f_stat) { + +void ThreadLocalStoreImpl::forEachTextReadout( + std::function f_size, + std::function f_stat) const { Thread::LockGuard lock(lock_); - alloc_.forEachSinkedTextReadout(f_size, f_stat); + alloc_.forEachTextReadout(f_size, f_stat); } } // namespace Stats diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h index ad16081b63c97..62160e938526e 100644 --- a/source/common/stats/thread_local_store.h +++ b/source/common/stats/thread_local_store.h @@ -244,14 +244,14 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo std::vector textReadouts() const override; std::vector histograms() const override; - void forEachSinkedCounter(std::function f_size, - std::function f_stat) override; + void forEachCounter(std::function f_size, + std::function f_stat) const override; - void forEachSinkedGauge(std::function f_size, - std::function f_stat) override; + void forEachGauge(std::function f_size, + std::function f_stat) const override; - void forEachSinkedTextReadout(std::function f_size, - std::function f_stat) override; + void forEachTextReadout(std::function f_size, + std::function f_stat) const override; // Stats::StoreRoot void addSink(Sink& sink) override { timer_sinks_.push_back(sink); } @@ -492,6 +492,8 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo bool rejectsAll() const { return stats_matcher_->rejectsAll(); } template void removeRejectedStats(StatMapClass& map, StatListClass& list); + template + void removeRejectedStats(StatMapClass& map, std::function f_deletion); bool checkAndRememberRejection(StatName name, StatsMatcher::FastResult fast_reject_result, StatNameStorageSet& central_rejected_stats, StatNameHashSet* tls_rejected_stats); @@ -536,10 +538,7 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo // It seems like it would be better to have each client that expects a stat // to exist to hold it as (e.g.) a CounterSharedPtr rather than a Counter& // but that would be fairly complex to change. - std::vector deleted_counters_ ABSL_GUARDED_BY(lock_); - std::vector deleted_gauges_ ABSL_GUARDED_BY(lock_); std::vector deleted_histograms_ ABSL_GUARDED_BY(lock_); - std::vector deleted_text_readouts_ ABSL_GUARDED_BY(lock_); // Scope IDs and central cache entries that are queued for cross-scope release. // Because there can be a large number of scopes, all of which are released at once, diff --git a/source/server/server.cc b/source/server/server.cc index daf3e6dc5c540..152058301be54 100644 --- a/source/server/server.cc +++ b/source/server/server.cc @@ -166,16 +166,16 @@ void InstanceImpl::failHealthcheck(bool fail) { } MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_source) { - store.forEachSinkedCounter([this](std::size_t size) mutable { counters_.reserve(size); }, - [this](Stats::Counter& counter) mutable { - counters_.push_back({counter.latch(), counter}); - }); + store.forEachCounter([this](std::size_t size) mutable { counters_.reserve(size); }, + [this](Stats::Counter& counter) mutable { + counters_.push_back({counter.latch(), counter}); + }); - store.forEachSinkedGauge([this](std::size_t size) mutable { gauges_.reserve(size); }, - [this](Stats::Gauge& gauge) mutable { - ASSERT(gauge.importMode() != Stats::Gauge::ImportMode::Uninitialized); - gauges_.push_back(gauge); - }); + store.forEachGauge([this](std::size_t size) mutable { gauges_.reserve(size); }, + [this](Stats::Gauge& gauge) mutable { + ASSERT(gauge.importMode() != Stats::Gauge::ImportMode::Uninitialized); + gauges_.push_back(gauge); + }); snapped_histograms_ = store.histograms(); histograms_.reserve(snapped_histograms_.size()); @@ -183,7 +183,7 @@ MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_sou histograms_.push_back(*histogram); } - store.forEachSinkedTextReadout( + store.forEachTextReadout( [this](std::size_t size) mutable { text_readouts_.reserve(size); }, [this](Stats::TextReadout& text_readout) { text_readouts_.push_back(text_readout); }); diff --git a/test/common/network/apple_dns_impl_test.cc b/test/common/network/apple_dns_impl_test.cc index feb6b0ac89dda..8b0e97ad75072 100644 --- a/test/common/network/apple_dns_impl_test.cc +++ b/test/common/network/apple_dns_impl_test.cc @@ -804,7 +804,7 @@ TEST_F(AppleDnsImplFakeApiTest, ResultWithNullAddress) { auto query = resolver_->resolve( hostname, Network::DnsLookupFamily::Auto, - [](DnsResolver::ResolutionStatus, std::list&&) -> void { FAIL(); }); + [](DnsResolver::ResolutionStatus, std::list &&) -> void { FAIL(); }); ASSERT_NE(nullptr, query); EXPECT_DEATH(reply_callback(nullptr, kDNSServiceFlagsAdd, 0, kDNSServiceErr_NoError, diff --git a/test/common/stats/thread_local_store_test.cc b/test/common/stats/thread_local_store_test.cc index b723fe3083fb9..84105391f71c5 100644 --- a/test/common/stats/thread_local_store_test.cc +++ b/test/common/stats/thread_local_store_test.cc @@ -508,10 +508,11 @@ TEST_F(StatsThreadLocalStoreTest, ScopeDelete) { EXPECT_CALL(main_thread_dispatcher_, post(_)); EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1)); scope1.reset(); - EXPECT_EQ(0UL, store_->counters().size()); + EXPECT_EQ(1UL, store_->counters().size()); EXPECT_EQ(1L, c1.use_count()); c1.reset(); + EXPECT_EQ(0UL, store_->counters().size()); tls_.shutdownGlobalThreading(); store_->shutdownThreading(); diff --git a/test/mocks/stats/mocks.h b/test/mocks/stats/mocks.h index 85894a7144b4c..ba81b5922f306 100644 --- a/test/mocks/stats/mocks.h +++ b/test/mocks/stats/mocks.h @@ -286,12 +286,13 @@ class MockStore : public TestUtil::TestStore { MOCK_METHOD(Histogram&, histogramFromString, (const std::string& name, Histogram::Unit unit)); MOCK_METHOD(TextReadout&, textReadout, (const std::string&)); MOCK_METHOD(std::vector, text_readouts, (), (const)); - MOCK_METHOD(void, forEachSinkedCounter, - (std::function, std::function)); - MOCK_METHOD(void, forEachSinkedGauge, - (std::function, std::function)); - MOCK_METHOD(void, forEachSinkedTextReadout, - (std::function, std::function)); + MOCK_METHOD(void, forEachCounter, + (std::function, std::function), (const)); + MOCK_METHOD(void, forEachGauge, + (std::function, std::function), (const)); + MOCK_METHOD(void, forEachTextReadout, + (std::function, std::function), + (const)); MOCK_METHOD(CounterOptConstRef, findCounter, (StatName), (const)); MOCK_METHOD(GaugeOptConstRef, findGauge, (StatName), (const)); diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index 34e6c57171517..250ea08122d59 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -18,13 +18,13 @@ #include "gtest/gtest.h" namespace Envoy { + class StatsSinkFlushSpeedTest { public: StatsSinkFlushSpeedTest(size_t const num_stats) : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { - sinks_.emplace_back(new testing::NiceMock()); // Create counters for (uint64_t idx = 0; idx < num_stats; ++idx) { auto stat_name = pool_.add(absl::StrCat("counter.", idx)); @@ -46,7 +46,9 @@ class StatsSinkFlushSpeedTest { void test(::benchmark::State& state) { for (auto _ : state) { UNREFERENCED_PARAMETER(state); - Server::InstanceUtil::flushMetricsToSinks(sinks_, stats_store_, time_system_); + std::list sinks; + sinks.emplace_back(new testing::NiceMock()); + Server::InstanceUtil::flushMetricsToSinks(sinks, stats_store_, time_system_); } } @@ -55,7 +57,6 @@ class StatsSinkFlushSpeedTest { Stats::StatNamePool pool_; Stats::AllocatorImpl stats_allocator_; Stats::ThreadLocalStoreImpl stats_store_; - std::list sinks_; Event::SimulatedTimeSystem time_system_; }; From e5a8a82fcc72900dd800665105fa6bf3dda89a7b Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Wed, 1 Sep 2021 12:55:11 +0000 Subject: [PATCH 08/12] Remove hasStat. Move deleted stats to stats set in ~AllocatorImpl. Signed-off-by: Pradeep Rao --- source/common/stats/allocator_impl.cc | 42 +++++++++++++++------------ 1 file changed, 23 insertions(+), 19 deletions(-) diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index ff093788c8ca3..240e6e7f39453 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -17,19 +17,6 @@ #include "absl/container/flat_hash_set.h" -namespace { -// Helper function to check if vector of StatSharedPtrs contains a StatPtr -// Note this is only meant to be used in asserts as this check is expensive. -template -bool hasStat(const std::vector>& vec, - StatType* const stat_ptr) { - return std::find_if(vec.begin(), vec.end(), - [stat_ptr](const Envoy::Stats::RefcountPtr& shared_ptr) { - return (shared_ptr.get() == stat_ptr); - }) != vec.end(); -} -} // namespace - namespace Envoy { namespace Stats { @@ -38,6 +25,23 @@ const char AllocatorImpl::DecrementToZeroSyncPoint[] = "decrement-zero"; AllocatorImpl::~AllocatorImpl() { ASSERT(counters_.empty()); ASSERT(gauges_.empty()); + + // Move deleted stats into the sets for removeFromSetLockHeld to function. + for (auto& counter : deleted_counters_) { + // Assert that there were no duplicates. + ASSERT(counters_.count(counter.get()) == 0); + counters_.insert(counter.get()); + } + for (auto& gauge : deleted_gauges_) { + // Assert that there were no duplicates. + ASSERT(gauges_.count(gauge.get()) == 0); + gauges_.insert(gauge.get()); + } + for (auto& text_readout : deleted_text_readouts_) { + // Assert that there were no duplicates. + ASSERT(text_readouts_.count(text_readout.get()) == 0); + text_readouts_.insert(text_readout.get()); + } } #ifndef ENVOY_CONFIG_COVERAGE @@ -136,7 +140,7 @@ class CounterImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.counters_.erase(statName()); - ASSERT(count == 1 || hasStat(alloc_.deleted_counters_, this)); + ASSERT(count == 1); } // Stats::Counter @@ -180,7 +184,7 @@ class GaugeImpl : public StatsSharedImpl { void removeFromSetLockHeld() override ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) { const size_t count = alloc_.gauges_.erase(statName()); - ASSERT(count == 1 || hasStat(alloc_.deleted_gauges_, this)); + ASSERT(count == 1); } // Stats::Gauge @@ -252,7 +256,7 @@ class TextReadoutImpl : public StatsSharedImpl { void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override { const size_t count = alloc_.text_readouts_.erase(statName()); - ASSERT(count == 1 || hasStat(alloc_.deleted_text_readouts_, this)); + ASSERT(count == 1); } // Stats::TextReadout @@ -362,7 +366,7 @@ void AllocatorImpl::markCounterForDeletion(StatName name) { if (iter == counters_.end()) { return; } - ASSERT(!hasStat(deleted_counters_, *iter)); + // Duplicates are checked in ~AllocatorImpl. deleted_counters_.emplace_back(*iter); counters_.erase(iter); } @@ -373,7 +377,7 @@ void AllocatorImpl::markGaugeForDeletion(StatName name) { if (iter == gauges_.end()) { return; } - ASSERT(!hasStat(deleted_gauges_, *iter)); + // Duplicates are checked in ~AllocatorImpl. deleted_gauges_.emplace_back(*iter); gauges_.erase(iter); } @@ -384,7 +388,7 @@ void AllocatorImpl::markTextReadoutForDeletion(StatName name) { if (iter == text_readouts_.end()) { return; } - ASSERT(!hasStat(deleted_text_readouts_, *iter)); + // Duplicates are checked in ~AllocatorImpl. deleted_text_readouts_.emplace_back(*iter); text_readouts_.erase(iter); } From d7b5ed50cb0460c5d93622d07e7ce264d23e8794 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Wed, 1 Sep 2021 20:52:50 +0000 Subject: [PATCH 09/12] Add tests for forEach apis. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 6 +- source/common/stats/allocator_impl.cc | 31 ++- source/common/stats/allocator_impl.h | 6 +- source/common/stats/thread_local_store.cc | 26 ++- source/common/stats/thread_local_store.h | 5 +- test/common/stats/allocator_impl_test.cc | 181 +++++++++++++++++- test/common/stats/thread_local_store_test.cc | 4 + .../server_stats_flush_benchmark_test.cc | 1 - tools/spelling/spelling_dictionary.txt | 2 +- 9 files changed, 233 insertions(+), 29 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index 3f5ad4112c5ad..b88b9972b23be 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -62,9 +62,9 @@ class Allocator { * Mark rejected stats as deleted by moving them to a different vector, so they don't show up * when iterating over stats, but prevent crashes when trying to accesses references to them. */ - virtual void markCounterForDeletion(StatName name) PURE; - virtual void markGaugeForDeletion(StatName name) PURE; - virtual void markTextReadoutForDeletion(StatName name) PURE; + virtual void markCounterForDeletion(const CounterSharedPtr& counter) PURE; + virtual void markGaugeForDeletion(const GaugeSharedPtr& gauge) PURE; + virtual void markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) PURE; /** * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index 35c088639d983..23434ff73e695 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -1,5 +1,6 @@ #include "source/common/stats/allocator_impl.h" +#include #include #include "envoy/stats/stats.h" @@ -347,9 +348,17 @@ void AllocatorImpl::forEachCounter(std::function f_size, void AllocatorImpl::forEachGauge(std::function f_size, std::function f_stat) const { Thread::LockGuard lock(mutex_); - f_size(gauges_.size()); + size_t num_gauges = 0; + std::for_each(gauges_.begin(), gauges_.end(), [&num_gauges](Gauge* const gauge) mutable { + if (gauge->importMode() != Gauge::ImportMode::Uninitialized) { + num_gauges++; + } + }); + f_size(num_gauges); for (auto& gauge : gauges_) { - f_stat(*gauge); + if (gauge->importMode() != Gauge::ImportMode::Uninitialized) { + f_stat(*gauge); + } } } @@ -362,34 +371,40 @@ void AllocatorImpl::forEachTextReadout(std::function f_size, } } -void AllocatorImpl::markCounterForDeletion(StatName name) { +void AllocatorImpl::markCounterForDeletion(const CounterSharedPtr& counter) { Thread::LockGuard lock(mutex_); - auto iter = counters_.find(name); + auto iter = counters_.find(counter->statName()); if (iter == counters_.end()) { + // This has already been marked for deletion. return; } + ASSERT(counter.get() == *iter); // Duplicates are ASSERTed in ~AllocatorImpl. deleted_counters_.emplace_back(*iter); counters_.erase(iter); } -void AllocatorImpl::markGaugeForDeletion(StatName name) { +void AllocatorImpl::markGaugeForDeletion(const GaugeSharedPtr& gauge) { Thread::LockGuard lock(mutex_); - auto iter = gauges_.find(name); + auto iter = gauges_.find(gauge->statName()); if (iter == gauges_.end()) { + // This has already been marked for deletion. return; } + ASSERT(gauge.get() == *iter); // Duplicates are ASSERTed in ~AllocatorImpl. deleted_gauges_.emplace_back(*iter); gauges_.erase(iter); } -void AllocatorImpl::markTextReadoutForDeletion(StatName name) { +void AllocatorImpl::markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) { Thread::LockGuard lock(mutex_); - auto iter = text_readouts_.find(name); + auto iter = text_readouts_.find(text_readout->statName()); if (iter == text_readouts_.end()) { + // This has already been marked for deletion. return; } + ASSERT(text_readout.get() == *iter); // Duplicates are ASSERTed in ~AllocatorImpl. deleted_text_readouts_.emplace_back(*iter); text_readouts_.erase(iter); diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h index c0a62eac16770..8c403e21750d9 100644 --- a/source/common/stats/allocator_impl.h +++ b/source/common/stats/allocator_impl.h @@ -57,9 +57,9 @@ class AllocatorImpl : public Allocator { */ bool isMutexLockedForTest(); - void markCounterForDeletion(StatName name) override; - void markGaugeForDeletion(StatName name) override; - void markTextReadoutForDeletion(StatName name) override; + void markCounterForDeletion(const CounterSharedPtr& counter) override; + void markGaugeForDeletion(const GaugeSharedPtr& gauge) override; + void markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) override; protected: virtual Counter* makeCounterInternal(StatName name, StatName tag_extracted_name, diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc index 0db5c94cb60df..1ae72e01f75c1 100644 --- a/source/common/stats/thread_local_store.cc +++ b/source/common/stats/thread_local_store.cc @@ -69,13 +69,19 @@ void ThreadLocalStoreImpl::setStatsMatcher(StatsMatcherPtr&& stats_matcher) { Thread::LockGuard lock(lock_); const uint32_t first_histogram_index = deleted_histograms_.size(); for (ScopeImpl* scope : scopes_) { - removeRejectedStats(scope->central_cache_->counters_, - [this](StatName name) mutable { alloc_.markCounterForDeletion(name); }); - removeRejectedStats(scope->central_cache_->gauges_, - [this](StatName name) mutable { alloc_.markGaugeForDeletion(name); }); + removeRejectedStats(scope->central_cache_->counters_, + [this](const CounterSharedPtr& counter) mutable { + alloc_.markCounterForDeletion(counter); + }); + removeRejectedStats( + scope->central_cache_->gauges_, + [this](const GaugeSharedPtr& gauge) mutable { alloc_.markGaugeForDeletion(gauge); }); removeRejectedStats(scope->central_cache_->histograms_, deleted_histograms_); - removeRejectedStats(scope->central_cache_->text_readouts_, - [this](StatName name) mutable { alloc_.markTextReadoutForDeletion(name); }); + removeRejectedStats( + scope->central_cache_->text_readouts_, + [this](const TextReadoutSharedPtr& text_readout) mutable { + alloc_.markTextReadoutForDeletion(text_readout); + }); } // Remove any newly rejected histograms from histogram_set_. @@ -104,9 +110,9 @@ void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, StatListClass& } } -template -void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, - std::function f_deletion) { +template +void ThreadLocalStoreImpl::removeRejectedStats( + StatNameHashMap& map, std::function f_deletion) { StatNameVec remove_list; for (auto& stat : map) { if (rejects(stat.first)) { @@ -116,7 +122,7 @@ void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, for (StatName stat_name : remove_list) { auto iter = map.find(stat_name); ASSERT(iter != map.end()); - f_deletion(stat_name); + f_deletion(iter->second); map.erase(iter); } } diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h index 62160e938526e..742e1fc3c04d6 100644 --- a/source/common/stats/thread_local_store.h +++ b/source/common/stats/thread_local_store.h @@ -492,8 +492,9 @@ class ThreadLocalStoreImpl : Logger::Loggable, public StoreRo bool rejectsAll() const { return stats_matcher_->rejectsAll(); } template void removeRejectedStats(StatMapClass& map, StatListClass& list); - template - void removeRejectedStats(StatMapClass& map, std::function f_deletion); + template + void removeRejectedStats(StatNameHashMap& map, + std::function f_deletion); bool checkAndRememberRejection(StatName name, StatsMatcher::FastResult fast_reject_result, StatNameStorageSet& central_rejected_stats, StatNameHashSet* tls_rejected_stats); diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index 83fb85d68ea8d..9f80dc0a3625d 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -1,3 +1,4 @@ +#include #include #include "source/common/stats/allocator_impl.h" @@ -6,6 +7,7 @@ #include "test/test_common/thread_factory_for_test.h" #include "absl/synchronization/notification.h" +#include "gmock/gmock-matchers.h" #include "gtest/gtest.h" namespace Envoy { @@ -25,12 +27,18 @@ class AllocatorImplTest : public testing::Test { void clearStorage() { pool_.clear(); - EXPECT_EQ(0, symbol_table_.numSymbols()); + // If stats have been marked for deletion, they are not cleared until the + // destructor of alloc_ is called, and hence the symbol_table_.numSymbols() + // will be greater than zero at this point. + if (are_stats_marked_for_deletion_ == false) { + EXPECT_EQ(0, symbol_table_.numSymbols()); + } } SymbolTableImpl symbol_table_; AllocatorImpl alloc_; StatNamePool pool_; + bool are_stats_marked_for_deletion_ = false; }; // Allocate 2 counters of the same name, and you'll get the same object. @@ -125,6 +133,177 @@ TEST_F(AllocatorImplTest, RefCountDecAllocRaceSynchronized) { EXPECT_FALSE(alloc_.isMutexLockedForTest()); } +TEST_F(AllocatorImplTest, ForEachCounter) { + + StatNameHashSet stat_names; + std::vector counters; + + const size_t num_stats = 11; + + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("counter.", idx)); + stat_names.insert(stat_name); + counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); + } + + size_t num_counters = 0; + size_t num_iterations = 0; + alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; }, + [&num_iterations, &stat_names](Stats::Counter& counter) { + EXPECT_EQ(stat_names.count(counter.statName()), 1); + ++num_iterations; + }); + EXPECT_EQ(num_counters, 11); + EXPECT_EQ(num_iterations, 11); + + // Reject a stat and remove it from "scope". + StatName rejected_stat_name = counters[4]->statName(); + alloc_.markCounterForDeletion(counters[4]); + are_stats_marked_for_deletion_ = true; + // Save a local reference to rejected stat. + Counter& rejected_counter = *counters[4]; + counters.erase(counters.begin() + 4); + + // Verify that the rejected stat does not show up during iteration. + num_iterations = 0; + num_counters = 0; + alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; }, + [&num_iterations, &rejected_stat_name](Stats::Counter& counter) { + EXPECT_THAT(counter.statName(), ::testing::Ne(rejected_stat_name)); + ++num_iterations; + }); + EXPECT_EQ(num_iterations, 10); + EXPECT_EQ(num_counters, 10); + + // Verify that we can access the local reference without a crash. + rejected_counter.inc(); + + // Erase all stats. + counters.clear(); + num_iterations = 0; + alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; }, + [&num_iterations](Stats::Counter&) { ++num_iterations; }); + EXPECT_EQ(num_counters, 0); + EXPECT_EQ(num_iterations, 0); +} + +TEST_F(AllocatorImplTest, ForEachGauge) { + + StatNameHashSet stat_names; + std::vector gauges; + + const size_t num_stats = 11; + + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("gauge.", idx)); + // Set every 5th gauge as Uninitialized. + if ((idx + 1) % 5 == 0) { + gauges.emplace_back( + alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Uninitialized)); + } else { + stat_names.insert(stat_name); + gauges.emplace_back( + alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); + } + } + + size_t num_gauges = 0; + size_t num_iterations = 0; + alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; }, + [&num_iterations, &stat_names](Stats::Gauge& gauge) { + EXPECT_EQ(stat_names.count(gauge.statName()), 1); + ++num_iterations; + }); + // We should see two less as we should not iterate over Uninitialized gauges. + EXPECT_EQ(num_gauges, 9); + EXPECT_EQ(num_iterations, 9); + + // Reject a stat and remove it from "scope". + StatName rejected_stat_name = gauges[3]->statName(); + alloc_.markGaugeForDeletion(gauges[3]); + are_stats_marked_for_deletion_ = true; + // Save a local reference to rejected stat. + Gauge& rejected_gauge = *gauges[3]; + gauges.erase(gauges.begin() + 3); + + // Verify that the rejected stat does not show up during iteration. + num_iterations = 0; + num_gauges = 0; + alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; }, + [&num_iterations, &rejected_stat_name](Stats::Gauge& gauge) { + EXPECT_THAT(gauge.statName(), ::testing::Ne(rejected_stat_name)); + ++num_iterations; + }); + EXPECT_EQ(num_iterations, 8); + EXPECT_EQ(num_gauges, 8); + + // Verify that we can access the local reference without a crash. + rejected_gauge.inc(); + + // Erase all stats. + gauges.clear(); + num_iterations = 0; + alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; }, + [&num_iterations](Stats::Gauge&) { ++num_iterations; }); + EXPECT_EQ(num_gauges, 0); + EXPECT_EQ(num_iterations, 0); +} + +TEST_F(AllocatorImplTest, ForEachTextReadout) { + + StatNameHashSet stat_names; + std::vector text_readouts; + + const size_t num_stats = 11; + + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("text_readout.", idx)); + stat_names.insert(stat_name); + text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); + } + + size_t num_text_readouts = 0; + size_t num_iterations = 0; + alloc_.forEachTextReadout([&num_text_readouts](std::size_t size) { num_text_readouts = size; }, + [&num_iterations, &stat_names](Stats::TextReadout& text_readout) { + EXPECT_EQ(stat_names.count(text_readout.statName()), 1); + ++num_iterations; + }); + EXPECT_EQ(num_text_readouts, 11); + EXPECT_EQ(num_iterations, 11); + + // Reject a stat and remove it from "scope". + StatName rejected_stat_name = text_readouts[4]->statName(); + alloc_.markTextReadoutForDeletion(text_readouts[4]); + are_stats_marked_for_deletion_ = true; + // Save a local reference to rejected stat. + TextReadout& rejected_text_readout = *text_readouts[4]; + text_readouts.erase(text_readouts.begin() + 4); + + // Verify that the rejected stat does not show up during iteration. + num_iterations = 0; + num_text_readouts = 0; + alloc_.forEachTextReadout( + [&num_text_readouts](std::size_t size) { num_text_readouts = size; }, + [&num_iterations, &rejected_stat_name](Stats::TextReadout& text_readout) { + EXPECT_THAT(text_readout.statName(), ::testing::Ne(rejected_stat_name)); + ++num_iterations; + }); + EXPECT_EQ(num_iterations, 10); + EXPECT_EQ(num_text_readouts, 10); + + // Verify that we can access the local reference without a crash. + rejected_text_readout.set("no crash"); + + // Erase all stats. + text_readouts.clear(); + num_iterations = 0; + alloc_.forEachTextReadout([&num_text_readouts](std::size_t size) { num_text_readouts = size; }, + [&num_iterations](Stats::TextReadout&) { ++num_iterations; }); + EXPECT_EQ(num_text_readouts, 0); + EXPECT_EQ(num_iterations, 0); +} + } // namespace } // namespace Stats } // namespace Envoy diff --git a/test/common/stats/thread_local_store_test.cc b/test/common/stats/thread_local_store_test.cc index 5bbb16478d7cc..28d7665bfa6cc 100644 --- a/test/common/stats/thread_local_store_test.cc +++ b/test/common/stats/thread_local_store_test.cc @@ -508,10 +508,14 @@ TEST_F(StatsThreadLocalStoreTest, ScopeDelete) { EXPECT_CALL(main_thread_dispatcher_, post(_)); EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1)); scope1.reset(); + // The counter is gone from all scopes, but is still held in the local + // variable c1. Hence, it will no be removed from the allocator or store. EXPECT_EQ(1UL, store_->counters().size()); EXPECT_EQ(1L, c1.use_count()); c1.reset(); + // Removing the counter from the local variable, should now remove it from the + // allocator. EXPECT_EQ(0UL, store_->counters().size()); tls_.shutdownGlobalThreading(); diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc index 250ea08122d59..6bb0f274e63a8 100644 --- a/test/server/server_stats_flush_benchmark_test.cc +++ b/test/server/server_stats_flush_benchmark_test.cc @@ -20,7 +20,6 @@ namespace Envoy { class StatsSinkFlushSpeedTest { - public: StatsSinkFlushSpeedTest(size_t const num_stats) : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) { diff --git a/tools/spelling/spelling_dictionary.txt b/tools/spelling/spelling_dictionary.txt index 569d0d5fa2e54..01c1a4a0403a2 100644 --- a/tools/spelling/spelling_dictionary.txt +++ b/tools/spelling/spelling_dictionary.txt @@ -15,7 +15,7 @@ ARN ASAN ASCII ASM -ASSER +ASSERTed ASSERTs AST AWS From 7cad62145d49291c08e6caf8977d41391c498e58 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Thu, 2 Sep 2021 14:56:21 +0000 Subject: [PATCH 10/12] Add test, fix crash. Signed-off-by: Pradeep Rao --- source/common/stats/allocator_impl.cc | 12 +++---- source/common/stats/allocator_impl.h | 14 +++----- test/common/stats/allocator_impl_test.cc | 46 ++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 16 deletions(-) diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc index 9783f39de80b8..4464f41a344e6 100644 --- a/source/common/stats/allocator_impl.cc +++ b/source/common/stats/allocator_impl.cc @@ -27,6 +27,7 @@ AllocatorImpl::~AllocatorImpl() { ASSERT(counters_.empty()); ASSERT(gauges_.empty()); +#ifndef NDEBUG // Move deleted stats into the sets for the ASSERTs in removeFromSetLockHeld to function. for (auto& counter : deleted_counters_) { auto insertion = counters_.insert(counter.get()); @@ -43,12 +44,7 @@ AllocatorImpl::~AllocatorImpl() { // Assert that there were no duplicates. ASSERT(insertion.second); } - // Make sure all stats are deleted BEFORE threads are shutdown. This is - // because the stat destructors need to lock a mutex, which will lead to - // undefined behavior once the threads are destroyed. - deleted_counters_.clear(); - deleted_text_readouts_.clear(); - deleted_gauges_.clear(); +#endif } #ifndef ENVOY_CONFIG_COVERAGE @@ -365,7 +361,9 @@ void AllocatorImpl::forEachGauge(std::function f_size, void AllocatorImpl::forEachTextReadout(std::function f_size, std::function f_stat) const { Thread::LockGuard lock(mutex_); - f_size(text_readouts_.size()); + if (f_size != nullptr) { + f_size(text_readouts_.size()); + } for (auto& text_readout : text_readouts_) { f_stat(*text_readout); } diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h index a7e8aa9991d7b..806e7dc8612fd 100644 --- a/source/common/stats/allocator_impl.h +++ b/source/common/stats/allocator_impl.h @@ -71,9 +71,11 @@ class AllocatorImpl : public Allocator { friend class TextReadoutImpl; friend class NotifyingAllocatorImpl; - // We don't need to check StatName to compare flushed stats. - template - using SinkedStatsSet = absl::flat_hash_set>; + // A mutex is needed here to protect both the stats_ object from both + // alloc() and free() operations. Although alloc() operations are called under existing locking, + // free() operations are made from the destructors of the individual stat objects, which are not + // protected by locks. + mutable Thread::MutexBasicLockable mutex_; StatSet counters_ ABSL_GUARDED_BY(mutex_); StatSet gauges_ ABSL_GUARDED_BY(mutex_); @@ -93,12 +95,6 @@ class AllocatorImpl : public Allocator { SymbolTable& symbol_table_; - // A mutex is needed here to protect both the stats_ object from both - // alloc() and free() operations. Although alloc() operations are called under existing locking, - // free() operations are made from the destructors of the individual stat objects, which are not - // protected by locks. - mutable Thread::MutexBasicLockable mutex_; - Thread::ThreadSynchronizer sync_; }; diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index 4a8352b55d4dc..d00c790e8e310 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -293,6 +293,52 @@ TEST_F(AllocatorImplTest, ForEachTextReadout) { EXPECT_EQ(num_iterations, 0); } +// Verify that we don't crash if a nullptr is passed in for the size lambda for +// the for each stat methods. +TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) { + std::vector counters; + std::vector text_readouts; + std::vector gauges; + + const size_t num_stats = 3; + + // For each counter. + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("counter.", idx)); + counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); + } + size_t num_iterations = 0; + alloc_.forEachCounter(nullptr, [&num_iterations](Stats::Counter& counter) { + (void)counter; + ++num_iterations; + }); + EXPECT_EQ(num_iterations, num_stats); + + // For each gauge. + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("gauge.", idx)); + gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); + } + num_iterations = 0; + alloc_.forEachGauge(nullptr, [&num_iterations](Stats::Gauge& gauge) { + (void)gauge; + ++num_iterations; + }); + EXPECT_EQ(num_iterations, num_stats); + + // For each text readout. + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("text_readout.", idx)); + text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); + } + num_iterations = 0; + alloc_.forEachTextReadout(nullptr, [&num_iterations](Stats::TextReadout& text_readout) { + (void)text_readout; + ++num_iterations; + }); + EXPECT_EQ(num_iterations, num_stats); +} + } // namespace } // namespace Stats } // namespace Envoy From 0f9ea87d559c357699578feaacc97cb9f10fe174 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Fri, 3 Sep 2021 18:07:27 +0000 Subject: [PATCH 11/12] Add test to document behavior for rejected stats on store and allocator. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 2 +- test/common/stats/allocator_impl_test.cc | 71 ++++++++++++++++++++ test/common/stats/thread_local_store_test.cc | 47 +++++++++++++ 3 files changed, 119 insertions(+), 1 deletion(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index b88b9972b23be..a1aa6c43db5e8 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -60,7 +60,7 @@ class Allocator { /** * Mark rejected stats as deleted by moving them to a different vector, so they don't show up - * when iterating over stats, but prevent crashes when trying to accesses references to them. + * when iterating over stats, but prevent crashes when trying to access references to them. */ virtual void markCounterForDeletion(const CounterSharedPtr& counter) PURE; virtual void markGaugeForDeletion(const GaugeSharedPtr& gauge) PURE; diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index d00c790e8e310..c3b5f7379ca15 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -339,6 +339,77 @@ TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) { EXPECT_EQ(num_iterations, num_stats); } +// Currently, if we ask for a stat from the Allocator that has already been +// marked for deletion (i.e. rejected) we get a new stat with the same name. +// This test documents this behavior. +TEST_F(AllocatorImplTest, AskForDeletedStat) { + const size_t num_stats = 10; + are_stats_marked_for_deletion_ = true; + + std::vector counters; + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("counter.", idx)); + counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {})); + } + // Reject a stat and remove it from "scope". + StatName const rejected_counter_name = counters[4]->statName(); + alloc_.markCounterForDeletion(counters[4]); + // Save a local reference to rejected stat. + Counter& rejected_counter = *counters[4]; + counters.erase(counters.begin() + 4); + + rejected_counter.inc(); + rejected_counter.inc(); + + // Make the deleted stat again. + CounterSharedPtr deleted_counter = alloc_.makeCounter(rejected_counter_name, StatName(), {}); + + EXPECT_EQ(deleted_counter->value(), 0); + EXPECT_EQ(rejected_counter.value(), 2); + + std::vector gauges; + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("gauge.", idx)); + gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate)); + } + // Reject a stat and remove it from "scope". + StatName const rejected_gauge_name = gauges[4]->statName(); + alloc_.markGaugeForDeletion(gauges[4]); + // Save a local reference to rejected stat. + Gauge& rejected_gauge = *gauges[4]; + gauges.erase(gauges.begin() + 4); + + rejected_gauge.set(10); + + // Make the deleted stat again. + GaugeSharedPtr deleted_gauge = + alloc_.makeGauge(rejected_gauge_name, StatName(), {}, Gauge::ImportMode::Accumulate); + + EXPECT_EQ(deleted_gauge->value(), 0); + EXPECT_EQ(rejected_gauge.value(), 10); + + std::vector text_readouts; + for (size_t idx = 0; idx < num_stats; ++idx) { + auto stat_name = makeStat(absl::StrCat("text_readout.", idx)); + text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {})); + } + // Reject a stat and remove it from "scope". + StatName const rejected_text_readout_name = text_readouts[4]->statName(); + alloc_.markTextReadoutForDeletion(text_readouts[4]); + // Save a local reference to rejected stat. + TextReadout& rejected_text_readout = *text_readouts[4]; + text_readouts.erase(text_readouts.begin() + 4); + + rejected_text_readout.set("deleted value"); + + // Make the deleted stat again. + TextReadoutSharedPtr deleted_text_readout = + alloc_.makeTextReadout(rejected_text_readout_name, StatName(), {}); + + EXPECT_EQ(deleted_text_readout->value(), ""); + EXPECT_EQ(rejected_text_readout.value(), "deleted value"); +} + } // namespace } // namespace Stats } // namespace Envoy diff --git a/test/common/stats/thread_local_store_test.cc b/test/common/stats/thread_local_store_test.cc index 50e32fe3981e2..907c5e1c58c6e 100644 --- a/test/common/stats/thread_local_store_test.cc +++ b/test/common/stats/thread_local_store_test.cc @@ -1197,6 +1197,53 @@ TEST_F(StatsThreadLocalStoreTest, RemoveRejectedStats) { tls_.shutdownThread(); } +// Verify that asking for deleted stats by name does not create new copies on +// the allocator. +TEST_F(StatsThreadLocalStoreTest, AskForRejectedStat) { + store_->initializeThreading(main_thread_dispatcher_, tls_); + Counter& counter = store_->counterFromString("c1"); + Gauge& gauge = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate); + TextReadout& textReadout = store_->textReadoutFromString("t1"); + ASSERT_EQ(1, store_->counters().size()); // "c1". + ASSERT_EQ(1, store_->gauges().size()); + ASSERT_EQ(1, store_->textReadouts().size()); + + // Will effectively block all stats, and remove all the non-matching stats. + envoy::config::metrics::v3::StatsConfig stats_config; + stats_config.mutable_stats_matcher()->mutable_inclusion_list()->add_patterns()->set_exact( + "no-such-stat"); + store_->setStatsMatcher(std::make_unique(stats_config, symbol_table_)); + + // They can no longer be found. + EXPECT_EQ(0, store_->counters().size()); + EXPECT_EQ(0, store_->gauges().size()); + EXPECT_EQ(0, store_->textReadouts().size()); + + // Set values for the rejected stats. + counter.inc(); + gauge.inc(); + textReadout.set("fortytwo"); + + // Ask for the rejected stats again by name. + Counter& counter2 = store_->counterFromString("c1"); + Gauge& gauge2 = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate); + TextReadout& textReadout2 = store_->textReadoutFromString("t1"); + + // Verify that we're getting the values previously set. + EXPECT_EQ(counter2.value(), 1); + EXPECT_EQ(gauge2.value(), 1); + EXPECT_EQ(textReadout2.value(), "fortytwo"); + + // Verify that new stats were not created. + EXPECT_EQ(0, store_->counters().size()); + EXPECT_EQ(0, store_->gauges().size()); + EXPECT_EQ(0, store_->textReadouts().size()); + + tls_.shutdownGlobalThreading(); + store_->shutdownThreading(); + tls_.shutdownThread(); +} + TEST_F(StatsThreadLocalStoreTest, NonHotRestartNoTruncation) { InSequence s; store_->initializeThreading(main_thread_dispatcher_, tls_); From 263205ee42df7ee74c3980c33363790371ac9c55 Mon Sep 17 00:00:00 2001 From: Pradeep Rao Date: Fri, 3 Sep 2021 19:12:03 +0000 Subject: [PATCH 12/12] Address comments. Signed-off-by: Pradeep Rao --- envoy/stats/allocator.h | 3 +++ test/common/stats/allocator_impl_test.cc | 2 +- test/common/stats/thread_local_store_test.cc | 17 ++++++----------- 3 files changed, 10 insertions(+), 12 deletions(-) diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h index a1aa6c43db5e8..6f9cc9715ea43 100644 --- a/envoy/stats/allocator.h +++ b/envoy/stats/allocator.h @@ -61,6 +61,9 @@ class Allocator { /** * Mark rejected stats as deleted by moving them to a different vector, so they don't show up * when iterating over stats, but prevent crashes when trying to access references to them. + * Note that allocating a stat with the same name after calling this will + * return a new stat. Hence callers should seek to avoid this situation, as is + * done in ThreadLocalStore. */ virtual void markCounterForDeletion(const CounterSharedPtr& counter) PURE; virtual void markGaugeForDeletion(const GaugeSharedPtr& gauge) PURE; diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc index c3b5f7379ca15..cc06acedaef1e 100644 --- a/test/common/stats/allocator_impl_test.cc +++ b/test/common/stats/allocator_impl_test.cc @@ -339,7 +339,7 @@ TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) { EXPECT_EQ(num_iterations, num_stats); } -// Currently, if we ask for a stat from the Allocator that has already been +// Currently, if we ask for a stat from the Allocator that has already been // marked for deletion (i.e. rejected) we get a new stat with the same name. // This test documents this behavior. TEST_F(AllocatorImplTest, AskForDeletedStat) { diff --git a/test/common/stats/thread_local_store_test.cc b/test/common/stats/thread_local_store_test.cc index 907c5e1c58c6e..5bf5e67395437 100644 --- a/test/common/stats/thread_local_store_test.cc +++ b/test/common/stats/thread_local_store_test.cc @@ -1203,7 +1203,7 @@ TEST_F(StatsThreadLocalStoreTest, AskForRejectedStat) { store_->initializeThreading(main_thread_dispatcher_, tls_); Counter& counter = store_->counterFromString("c1"); Gauge& gauge = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate); - TextReadout& textReadout = store_->textReadoutFromString("t1"); + TextReadout& text_readout = store_->textReadoutFromString("t1"); ASSERT_EQ(1, store_->counters().size()); // "c1". ASSERT_EQ(1, store_->gauges().size()); ASSERT_EQ(1, store_->textReadouts().size()); @@ -1219,20 +1219,15 @@ TEST_F(StatsThreadLocalStoreTest, AskForRejectedStat) { EXPECT_EQ(0, store_->gauges().size()); EXPECT_EQ(0, store_->textReadouts().size()); - // Set values for the rejected stats. - counter.inc(); - gauge.inc(); - textReadout.set("fortytwo"); - // Ask for the rejected stats again by name. Counter& counter2 = store_->counterFromString("c1"); Gauge& gauge2 = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate); - TextReadout& textReadout2 = store_->textReadoutFromString("t1"); + TextReadout& text_readout2 = store_->textReadoutFromString("t1"); - // Verify that we're getting the values previously set. - EXPECT_EQ(counter2.value(), 1); - EXPECT_EQ(gauge2.value(), 1); - EXPECT_EQ(textReadout2.value(), "fortytwo"); + // Verify we got the same stats. + EXPECT_EQ(&counter, &counter2); + EXPECT_EQ(&gauge, &gauge2); + EXPECT_EQ(&text_readout, &text_readout2); // Verify that new stats were not created. EXPECT_EQ(0, store_->counters().size());