From da333f8eba8270d1e34ebe8a3365485b473086a3 Mon Sep 17 00:00:00 2001 From: Lalit Kumar Bhasin Date: Fri, 3 Mar 2023 20:59:42 -0800 Subject: [PATCH] [Metrics SDK] Performance improvement in measurement processing (#1993) --- .../sdk/common/attributemap_hash.h | 42 +++++++-- .../sdk/metrics/state/async_metric_storage.h | 12 +-- .../sdk/metrics/state/attributes_hashmap.h | 89 +++++++++++++++---- .../sdk/metrics/state/sync_metric_storage.h | 43 ++++++--- .../sdk/metrics/view/attributes_processor.h | 17 +++- .../metrics/state/temporal_metric_storage.cc | 15 ++-- sdk/test/common/attributemap_hash_test.cc | 1 + sdk/test/metrics/BUILD | 16 ++++ sdk/test/metrics/CMakeLists.txt | 4 + .../metrics/attributes_hashmap_benchmark.cc | 6 +- sdk/test/metrics/attributes_hashmap_test.cc | 34 ++++--- sdk/test/metrics/measurements_benchmark.cc | 86 ++++++++++++++++++ 12 files changed, 300 insertions(+), 65 deletions(-) create mode 100644 sdk/test/metrics/measurements_benchmark.cc diff --git a/sdk/include/opentelemetry/sdk/common/attributemap_hash.h b/sdk/include/opentelemetry/sdk/common/attributemap_hash.h index c6f2c93ccc..086f6b7901 100644 --- a/sdk/include/opentelemetry/sdk/common/attributemap_hash.h +++ b/sdk/include/opentelemetry/sdk/common/attributemap_hash.h @@ -3,7 +3,6 @@ #pragma once -#include #include #include "opentelemetry/sdk/common/attribute_utils.h" @@ -14,7 +13,7 @@ namespace common { template -inline void GetHashForAttributeValue(size_t &seed, const T arg) +inline void GetHash(size_t &seed, const T &arg) { std::hash hasher; // reference - @@ -23,11 +22,11 @@ inline void GetHashForAttributeValue(size_t &seed, const T arg) } template -inline void GetHashForAttributeValue(size_t &seed, const std::vector &arg) +inline void GetHash(size_t &seed, const std::vector &arg) { for (auto v : arg) { - GetHashForAttributeValue(seed, v); + GetHash(seed, v); } } @@ -37,7 +36,7 @@ struct GetHashForAttributeValueVisitor template void operator()(T &v) { - GetHashForAttributeValue(seed_, v); + GetHash(seed_, v); } size_t &seed_; }; @@ -48,15 +47,40 @@ inline size_t GetHashForAttributeMap(const OrderedAttributeMap &attribute_map) size_t seed = 0UL; for (auto &kv : attribute_map) { - std::hash hasher; - // reference - - // https://www.boost.org/doc/libs/1_37_0/doc/html/hash/reference.html#boost.hash_combine - seed ^= hasher(kv.first) + 0x9e3779b9 + (seed << 6) + (seed >> 2); + GetHash(seed, kv.first); nostd::visit(GetHashForAttributeValueVisitor(seed), kv.second); } return seed; } +// Calculate hash of keys and values of KeyValueIterable, filtered using callback. +inline size_t GetHashForAttributeMap( + const opentelemetry::common::KeyValueIterable &attributes, + nostd::function_ref is_key_present_callback) +{ + AttributeConverter converter; + size_t seed = 0UL; + attributes.ForEachKeyValue( + [&](nostd::string_view key, opentelemetry::common::AttributeValue value) noexcept { + if (!is_key_present_callback(key)) + { + return true; + } + GetHash(seed, key.data()); + auto attr_val = nostd::visit(converter, value); + nostd::visit(GetHashForAttributeValueVisitor(seed), attr_val); + return true; + }); + return seed; +} + +template +inline size_t GetHash(T value) +{ + std::hash hasher; + return hasher(value); +} + } // namespace common } // namespace sdk OPENTELEMETRY_END_NAMESPACE diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h index f601691fd1..aad00bb5cf 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h @@ -47,22 +47,24 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora { auto aggr = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_); aggr->Aggregate(measurement.second); - auto prev = cumulative_hash_map_->Get(measurement.first); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(measurement.first); + auto prev = cumulative_hash_map_->Get(hash); if (prev) { auto delta = prev->Diff(*aggr); // store received value in cumulative map, and the diff in delta map (to pass it to temporal // storage) - cumulative_hash_map_->Set(measurement.first, std::move(aggr)); - delta_hash_map_->Set(measurement.first, std::move(delta)); + cumulative_hash_map_->Set(measurement.first, std::move(aggr), hash); + delta_hash_map_->Set(measurement.first, std::move(delta), hash); } else { // store received value in cumulative and delta map. cumulative_hash_map_->Set( measurement.first, - DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr)); - delta_hash_map_->Set(measurement.first, std::move(aggr)); + DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr), + hash); + delta_hash_map_->Set(measurement.first, std::move(aggr), hash); } } } diff --git a/sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h b/sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h index 5b575fd024..a4b53a37c5 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h @@ -8,6 +8,7 @@ #include "opentelemetry/sdk/common/attributemap_hash.h" #include "opentelemetry/sdk/metrics/aggregation/aggregation.h" #include "opentelemetry/sdk/metrics/instruments.h" +#include "opentelemetry/sdk/metrics/view/attributes_processor.h" #include "opentelemetry/version.h" #include @@ -19,6 +20,7 @@ namespace sdk { namespace metrics { + using opentelemetry::sdk::common::OrderedAttributeMap; class AttributeHashGenerator @@ -33,12 +35,12 @@ class AttributeHashGenerator class AttributesHashMap { public: - Aggregation *Get(const MetricAttributes &attributes) const + Aggregation *Get(size_t hash) const { - auto it = hash_map_.find(attributes); + auto it = hash_map_.find(hash); if (it != hash_map_.end()) { - return it->second.get(); + return it->second.second.get(); } return nullptr; } @@ -47,35 +49,89 @@ class AttributesHashMap * @return check if key is present in hash * */ - bool Has(const MetricAttributes &attributes) const - { - return (hash_map_.find(attributes) == hash_map_.end()) ? false : true; - } + bool Has(size_t hash) const { return hash_map_.find(hash) != hash_map_.end(); } /** * @return the pointer to value for given key if present. * If not present, it uses the provided callback to generate * value and store in the hash */ + Aggregation *GetOrSetDefault(const opentelemetry::common::KeyValueIterable &attributes, + std::function()> aggregation_callback, + size_t hash) + { + auto it = hash_map_.find(hash); + if (it != hash_map_.end()) + { + return it->second.second.get(); + } + + MetricAttributes attr{attributes}; + + hash_map_[hash] = {attr, aggregation_callback()}; + return hash_map_[hash].second.get(); + } + + Aggregation *GetOrSetDefault(std::function()> aggregation_callback, + size_t hash) + { + auto it = hash_map_.find(hash); + if (it != hash_map_.end()) + { + return it->second.second.get(); + } + MetricAttributes attr{}; + hash_map_[hash] = {attr, aggregation_callback()}; + return hash_map_[hash].second.get(); + } + Aggregation *GetOrSetDefault(const MetricAttributes &attributes, - std::function()> aggregation_callback) + std::function()> aggregation_callback, + size_t hash) { - auto it = hash_map_.find(attributes); + auto it = hash_map_.find(hash); if (it != hash_map_.end()) { - return it->second.get(); + return it->second.second.get(); } - hash_map_[attributes] = aggregation_callback(); - return hash_map_[attributes].get(); + MetricAttributes attr{attributes}; + + hash_map_[hash] = {attr, aggregation_callback()}; + return hash_map_[hash].second.get(); } /** * Set the value for given key, overwriting the value if already present */ - void Set(const MetricAttributes &attributes, std::unique_ptr value) + void Set(const opentelemetry::common::KeyValueIterable &attributes, + std::unique_ptr aggr, + size_t hash) { - hash_map_[attributes] = std::move(value); + auto it = hash_map_.find(hash); + if (it != hash_map_.end()) + { + it->second.second = std::move(aggr); + } + else + { + MetricAttributes attr{attributes}; + hash_map_[hash] = {attr, std::move(aggr)}; + } + } + + void Set(const MetricAttributes &attributes, std::unique_ptr aggr, size_t hash) + { + auto it = hash_map_.find(hash); + if (it != hash_map_.end()) + { + it->second.second = std::move(aggr); + } + else + { + MetricAttributes attr{attributes}; + hash_map_[hash] = {attr, std::move(aggr)}; + } } /** @@ -86,7 +142,7 @@ class AttributesHashMap { for (auto &kv : hash_map_) { - if (!callback(kv.first, *(kv.second.get()))) + if (!callback(kv.second.first, *(kv.second.second.get()))) { return false; // callback is not prepared to consume data } @@ -100,8 +156,7 @@ class AttributesHashMap size_t Size() { return hash_map_.size(); } private: - std::unordered_map, AttributeHashGenerator> - hash_map_; + std::unordered_map>> hash_map_; }; } // namespace metrics diff --git a/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h index da4fb2980a..b8146872e3 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h @@ -35,12 +35,11 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage const AggregationConfig *aggregation_config) : instrument_descriptor_(instrument_descriptor), attributes_hashmap_(new AttributesHashMap()), - attributes_processor_{attributes_processor}, + attributes_processor_(attributes_processor), #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW exemplar_reservoir_(exemplar_reservoir), #endif temporal_metric_storage_(instrument_descriptor, aggregation_type, aggregation_config) - { create_default_aggregation_ = [&, aggregation_type, aggregation_config]() -> std::unique_ptr { @@ -60,8 +59,9 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now()); #endif + static size_t hash = opentelemetry::sdk::common::GetHash(""); std::lock_guard guard(attribute_hashmap_lock_); - attributes_hashmap_->GetOrSetDefault({}, create_default_aggregation_)->Aggregate(value); + attributes_hashmap_->GetOrSetDefault(create_default_aggregation_, hash)->Aggregate(value); } void RecordLong(int64_t value, @@ -77,9 +77,21 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage exemplar_reservoir_->OfferMeasurement(value, attributes, context, std::chrono::system_clock::now()); #endif - auto attr = attributes_processor_->process(attributes); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap( + attributes, [this](nostd::string_view key) { + if (attributes_processor_) + { + return attributes_processor_->isPresent(key); + } + else + { + return true; + } + }); + std::lock_guard guard(attribute_hashmap_lock_); - attributes_hashmap_->GetOrSetDefault(attr, create_default_aggregation_)->Aggregate(value); + attributes_hashmap_->GetOrSetDefault(attributes, create_default_aggregation_, hash) + ->Aggregate(value); } void RecordDouble(double value, @@ -93,8 +105,9 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now()); #endif + static size_t hash = opentelemetry::sdk::common::GetHash(""); std::lock_guard guard(attribute_hashmap_lock_); - attributes_hashmap_->GetOrSetDefault({}, create_default_aggregation_)->Aggregate(value); + attributes_hashmap_->GetOrSetDefault(create_default_aggregation_, hash)->Aggregate(value); } void RecordDouble(double value, @@ -114,9 +127,20 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage exemplar_reservoir_->OfferMeasurement(value, attributes, context, std::chrono::system_clock::now()); #endif - auto attr = attributes_processor_->process(attributes); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap( + attributes, [this](nostd::string_view key) { + if (attributes_processor_) + { + return attributes_processor_->isPresent(key); + } + else + { + return true; + } + }); std::lock_guard guard(attribute_hashmap_lock_); - attributes_hashmap_->GetOrSetDefault(attr, create_default_aggregation_)->Aggregate(value); + attributes_hashmap_->GetOrSetDefault(attributes, create_default_aggregation_, hash) + ->Aggregate(value); } bool Collect(CollectorHandle *collector, @@ -127,7 +151,6 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage private: InstrumentDescriptor instrument_descriptor_; - // hashmap to maintain the metrics for delta collection (i.e, collection since last Collect call) std::unique_ptr attributes_hashmap_; // unreported metrics stash for all the collectors @@ -135,8 +158,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage unreported_metrics_; // last reported metrics stash for all the collectors. std::unordered_map last_reported_metrics_; - const AttributesProcessor *attributes_processor_; std::function()> create_default_aggregation_; + const AttributesProcessor *attributes_processor_; #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW nostd::shared_ptr exemplar_reservoir_; #endif diff --git a/sdk/include/opentelemetry/sdk/metrics/view/attributes_processor.h b/sdk/include/opentelemetry/sdk/metrics/view/attributes_processor.h index 754f45e1b6..c3328bfa1e 100644 --- a/sdk/include/opentelemetry/sdk/metrics/view/attributes_processor.h +++ b/sdk/include/opentelemetry/sdk/metrics/view/attributes_processor.h @@ -4,6 +4,7 @@ #pragma once #include "opentelemetry/sdk/common/attribute_utils.h" + OPENTELEMETRY_BEGIN_NAMESPACE namespace sdk { @@ -20,10 +21,14 @@ class AttributesProcessor { public: // Process the metric instrument attributes. - // @returns The processed attributes + // @returns integer with individual bits set if they are to be filtered. + virtual MetricAttributes process( const opentelemetry::common::KeyValueIterable &attributes) const noexcept = 0; - virtual ~AttributesProcessor() = default; + + virtual bool isPresent(nostd::string_view key) const noexcept = 0; + + virtual ~AttributesProcessor() = default; }; /** @@ -33,12 +38,15 @@ class AttributesProcessor class DefaultAttributesProcessor : public AttributesProcessor { +public: MetricAttributes process( const opentelemetry::common::KeyValueIterable &attributes) const noexcept override { MetricAttributes result(attributes); return result; } + + bool isPresent(nostd::string_view /*key*/) const noexcept override { return true; } }; /** @@ -70,6 +78,11 @@ class FilteringAttributesProcessor : public AttributesProcessor return result; } + bool isPresent(nostd::string_view key) const noexcept override + { + return (allowed_attribute_keys_.find(key.data()) != allowed_attribute_keys_.end()); + } + private: std::unordered_map allowed_attribute_keys_; }; diff --git a/sdk/src/metrics/state/temporal_metric_storage.cc b/sdk/src/metrics/state/temporal_metric_storage.cc index 50803a80c0..1095c21418 100644 --- a/sdk/src/metrics/state/temporal_metric_storage.cc +++ b/sdk/src/metrics/state/temporal_metric_storage.cc @@ -60,17 +60,19 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, { agg_hashmap->GetAllEnteries( [&merged_metrics, this](const MetricAttributes &attributes, Aggregation &aggregation) { - auto agg = merged_metrics->Get(attributes); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(attributes); + auto agg = merged_metrics->Get(hash); if (agg) { - merged_metrics->Set(attributes, agg->Merge(aggregation)); + merged_metrics->Set(attributes, agg->Merge(aggregation), hash); } else { merged_metrics->Set(attributes, DefaultAggregation::CreateAggregation( aggregation_type_, instrument_descriptor_, aggregation_config_) - ->Merge(aggregation)); + ->Merge(aggregation), + hash); } return true; }); @@ -94,16 +96,17 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, // merge current delta to previous cumulative last_aggr_hashmap->GetAllEnteries( [&merged_metrics, this](const MetricAttributes &attributes, Aggregation &aggregation) { - auto agg = merged_metrics->Get(attributes); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(attributes); + auto agg = merged_metrics->Get(hash); if (agg) { - merged_metrics->Set(attributes, agg->Merge(aggregation)); + merged_metrics->Set(attributes, agg->Merge(aggregation), hash); } else { auto def_agg = DefaultAggregation::CreateAggregation( aggregation_type_, instrument_descriptor_, aggregation_config_); - merged_metrics->Set(attributes, def_agg->Merge(aggregation)); + merged_metrics->Set(attributes, def_agg->Merge(aggregation), hash); } return true; }); diff --git a/sdk/test/common/attributemap_hash_test.cc b/sdk/test/common/attributemap_hash_test.cc index 49e53361ba..a7a893f2a8 100644 --- a/sdk/test/common/attributemap_hash_test.cc +++ b/sdk/test/common/attributemap_hash_test.cc @@ -19,6 +19,7 @@ TEST(AttributeMapHashTest, BasicTests) } { + // hash algo returns same value irrespective of order of attributes. OrderedAttributeMap map1 = {{"k1", 10}, {"k2", true}, {"k3", 12.22}}; OrderedAttributeMap map2 = {{"k3", 12.22}, {"k1", 10}, {"k2", true}}; EXPECT_TRUE(GetHashForAttributeMap(map1) == GetHashForAttributeMap(map2)); diff --git a/sdk/test/metrics/BUILD b/sdk/test/metrics/BUILD index 61c526bb0c..d7ed9e5fb6 100644 --- a/sdk/test/metrics/BUILD +++ b/sdk/test/metrics/BUILD @@ -363,3 +363,19 @@ otel_cc_benchmark( "//sdk/src/resource", ], ) + +otel_cc_benchmark( + name = "measurements_benchmark", + srcs = [ + "measurements_benchmark.cc", + ], + tags = [ + "benchmark", + "metrics", + "test", + ], + deps = [ + "//sdk/src/metrics", + "//sdk/src/resource", + ], +) diff --git a/sdk/test/metrics/CMakeLists.txt b/sdk/test/metrics/CMakeLists.txt index dc81b07e61..c167a27dd5 100644 --- a/sdk/test/metrics/CMakeLists.txt +++ b/sdk/test/metrics/CMakeLists.txt @@ -54,6 +54,10 @@ if(WITH_BENCHMARK) ${CMAKE_THREAD_LIBS_INIT} metrics_common_test_utils opentelemetry_common opentelemetry_resources) + add_executable(measurements_benchmark measurements_benchmark.cc) + target_link_libraries( + measurements_benchmark benchmark::benchmark opentelemetry_metrics + opentelemetry_resources ${CMAKE_THREAD_LIBS_INIT} opentelemetry_common) add_executable(sum_aggregation_benchmark sum_aggregation_benchmark.cc) target_link_libraries( sum_aggregation_benchmark benchmark::benchmark ${CMAKE_THREAD_LIBS_INIT} diff --git a/sdk/test/metrics/attributes_hashmap_benchmark.cc b/sdk/test/metrics/attributes_hashmap_benchmark.cc index 71aaa1cd6f..f0286ec09e 100644 --- a/sdk/test/metrics/attributes_hashmap_benchmark.cc +++ b/sdk/test/metrics/attributes_hashmap_benchmark.cc @@ -32,8 +32,10 @@ void BM_AttributseHashMap(benchmark::State &state) return std::unique_ptr(new DropAggregation); }; m.lock(); - hash_map.GetOrSetDefault(attributes[i % 2], create_default_aggregation)->Aggregate((int64_t)1); - benchmark::DoNotOptimize(hash_map.Has(attributes[i % 2])); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(attributes[i % 2]); + hash_map.GetOrSetDefault(attributes[i % 2], create_default_aggregation, hash) + ->Aggregate((int64_t)1); + benchmark::DoNotOptimize(hash_map.Has(hash)); m.unlock(); }; while (state.KeepRunning()) diff --git a/sdk/test/metrics/attributes_hashmap_test.cc b/sdk/test/metrics/attributes_hashmap_test.cc index dd74d5c75b..956a76dfe1 100644 --- a/sdk/test/metrics/attributes_hashmap_test.cc +++ b/sdk/test/metrics/attributes_hashmap_test.cc @@ -18,31 +18,34 @@ TEST(AttributesHashMap, BasicTests) AttributesHashMap hash_map; EXPECT_EQ(hash_map.Size(), 0); MetricAttributes m1 = {{"k1", "v1"}}; - EXPECT_EQ(hash_map.Get(m1), nullptr); - EXPECT_EQ(hash_map.Has(m1), false); + auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(m1); + + EXPECT_EQ(hash_map.Get(hash), nullptr); + EXPECT_EQ(hash_map.Has(hash), false); // Set std::unique_ptr aggregation1( new DropAggregation()); // = std::unique_ptr(new DropAggregation); - hash_map.Set(m1, std::move(aggregation1)); - EXPECT_NO_THROW(hash_map.Get(m1)->Aggregate((int64_t)1)); + hash_map.Set(m1, std::move(aggregation1), hash); + EXPECT_NO_THROW(hash_map.Get(hash)->Aggregate((int64_t)1)); EXPECT_EQ(hash_map.Size(), 1); - EXPECT_EQ(hash_map.Has(m1), true); + EXPECT_EQ(hash_map.Has(hash), true); // Set same key again auto aggregation2 = std::unique_ptr(new DropAggregation()); - hash_map.Set(m1, std::move(aggregation2)); - EXPECT_NO_THROW(hash_map.Get(m1)->Aggregate((int64_t)1)); + hash_map.Set(m1, std::move(aggregation2), hash); + EXPECT_NO_THROW(hash_map.Get(hash)->Aggregate((int64_t)1)); EXPECT_EQ(hash_map.Size(), 1); - EXPECT_EQ(hash_map.Has(m1), true); + EXPECT_EQ(hash_map.Has(hash), true); // Set more enteria auto aggregation3 = std::unique_ptr(new DropAggregation()); MetricAttributes m3 = {{"k1", "v1"}, {"k2", "v2"}}; - hash_map.Set(m3, std::move(aggregation3)); - EXPECT_EQ(hash_map.Has(m1), true); - EXPECT_EQ(hash_map.Has(m3), true); - EXPECT_NO_THROW(hash_map.Get(m3)->Aggregate((int64_t)1)); + auto hash3 = opentelemetry::sdk::common::GetHashForAttributeMap(m3); + hash_map.Set(m3, std::move(aggregation3), hash3); + EXPECT_EQ(hash_map.Has(hash), true); + EXPECT_EQ(hash_map.Has(hash3), true); + EXPECT_NO_THROW(hash_map.Get(hash3)->Aggregate((int64_t)1)); EXPECT_EQ(hash_map.Size(), 2); // GetOrSetDefault @@ -51,12 +54,15 @@ TEST(AttributesHashMap, BasicTests) return std::unique_ptr(new DropAggregation); }; MetricAttributes m4 = {{"k1", "v1"}, {"k2", "v2"}, {"k3", "v3"}}; - EXPECT_NO_THROW(hash_map.GetOrSetDefault(m4, create_default_aggregation)->Aggregate((int64_t)1)); + auto hash4 = opentelemetry::sdk::common::GetHashForAttributeMap(m4); + EXPECT_NO_THROW( + hash_map.GetOrSetDefault(m4, create_default_aggregation, hash4)->Aggregate((int64_t)1)); EXPECT_EQ(hash_map.Size(), 3); // Set attributes with different order - shouldn't create a new entry. MetricAttributes m5 = {{"k2", "v2"}, {"k1", "v1"}}; - EXPECT_EQ(hash_map.Has(m5), true); + auto hash5 = opentelemetry::sdk::common::GetHashForAttributeMap(m5); + EXPECT_EQ(hash_map.Has(hash5), true); // GetAllEnteries size_t count = 0; diff --git a/sdk/test/metrics/measurements_benchmark.cc b/sdk/test/metrics/measurements_benchmark.cc new file mode 100644 index 0000000000..06ac75a65a --- /dev/null +++ b/sdk/test/metrics/measurements_benchmark.cc @@ -0,0 +1,86 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#include "opentelemetry/sdk/metrics/meter.h" +#include "opentelemetry/sdk/metrics/meter_context.h" +#include "opentelemetry/sdk/metrics/meter_provider.h" +#include "opentelemetry/sdk/metrics/metric_reader.h" +#include "opentelemetry/sdk/metrics/push_metric_exporter.h" + +#include +#include +#include + +using namespace opentelemetry; +using namespace opentelemetry::sdk::instrumentationscope; +using namespace opentelemetry::sdk::metrics; +using namespace opentelemetry::sdk::common; + +class MockMetricExporter : public MetricReader +{ +public: + MockMetricExporter() = default; + + opentelemetry::sdk::metrics::AggregationTemporality GetAggregationTemporality( + opentelemetry::sdk::metrics::InstrumentType) const noexcept override + { + return AggregationTemporality::kCumulative; + } + +private: + bool OnForceFlush(std::chrono::microseconds /*timeout*/) noexcept override { return true; } + + bool OnShutDown(std::chrono::microseconds /*timeout*/) noexcept override { return true; } + + void OnInitialized() noexcept override {} +}; + +namespace +{ +void BM_MeasurementsTest(benchmark::State &state) +{ + MeterProvider mp; + auto m = mp.GetMeter("meter1", "version1", "schema1"); + + std::shared_ptr exporter(new MockMetricExporter()); + mp.AddMetricReader(exporter); + auto h = m->CreateDoubleCounter("counter1", "counter1_description", "counter1_unit"); + size_t MAX_MEASUREMENTS = 10000; // keep low to prevent CI failure due to timeout + size_t NUM_CORES = 1; + std::vector threads; + std::map attributes[1000]; + size_t total_index = 0; + for (uint32_t i = 0; i < 10; i++) + { + for (uint32_t j = 0; j < 10; j++) + for (uint32_t k = 0; k < 10; k++) + attributes[total_index++] = {{"dim1", i}, {"dim2", j}, {"dim3", k}}; + } + while (state.KeepRunning()) + { + threads.clear(); + std::atomic cur_processed{0}; + for (size_t i = 0; i < NUM_CORES; i++) + { + threads.push_back(std::thread([&h, &cur_processed, &MAX_MEASUREMENTS, &attributes]() { + while (cur_processed++ <= MAX_MEASUREMENTS) + { + size_t index = rand() % 1000; + h->Add(1.0, + opentelemetry::common::KeyValueIterableView>( + attributes[index]), + opentelemetry::context::Context{}); + } + })); + } + for (auto &thread : threads) + { + thread.join(); + } + } + exporter->Collect([&](ResourceMetrics & /*rm*/) { return true; }); +} +BENCHMARK(BM_MeasurementsTest); + +} // namespace +BENCHMARK_MAIN();