diff --git a/include/common/FeatureReport.h b/include/common/FeatureReport.h new file mode 100644 index 0000000..59f35ec --- /dev/null +++ b/include/common/FeatureReport.h @@ -0,0 +1,74 @@ +// Copyright (C) 2019-2020 Zilliz. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software distributed under the License +// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +// or implied. See the License for the specific language governing permissions and limitations under the License. + +#pragma once + +#include + +#include +#include +#include +#include + +namespace milvus::monitor::feature_report { + +#define MILVUS_FEATURE_REPORTERS(FEATURE) \ + FEATURE(HybridSearch, "hybrid_search") \ + FEATURE(PartitionKey, "partition_key") \ + FEATURE(DynamicField, "dynamic_field") \ + FEATURE(BM25Function, "bm25_function") \ + FEATURE(ResourceGroup, "resource_group") \ + FEATURE(BulkImport, "bulk_import") + +class alignas(64) FeatureReporter { + public: + FeatureReporter(const FeatureReporter&) = delete; + FeatureReporter& + operator=(const FeatureReporter&) = delete; + + std::string_view + Name() const noexcept; + + bool + Record(); + + private: + explicit FeatureReporter(std::string_view name); + + bool + recordAt(std::chrono::steady_clock::time_point now); + + void + reset(); + + friend struct FeatureReporterTestPeer; + +#define FRIEND_FEATURE_REPORTER(name, label) friend FeatureReporter& name(); + + MILVUS_FEATURE_REPORTERS(FRIEND_FEATURE_REPORTER) + +#undef FRIEND_FEATURE_REPORTER + + private: + std::atomic next_allowed_nanos_{0}; + std::string name_; + prometheus::Counter& counter_; +}; + +#define DECLARE_FEATURE_REPORTER(name, label) \ + constexpr std::string_view k##name = label; \ + FeatureReporter& name(); + +MILVUS_FEATURE_REPORTERS(DECLARE_FEATURE_REPORTER) + +#undef DECLARE_FEATURE_REPORTER + +} // namespace milvus::monitor::feature_report diff --git a/src/common/FeatureReport.cpp b/src/common/FeatureReport.cpp new file mode 100644 index 0000000..0cacd1c --- /dev/null +++ b/src/common/FeatureReport.cpp @@ -0,0 +1,95 @@ +// Copyright (C) 2019-2020 Zilliz. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software distributed under the License +// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +// or implied. See the License for the specific language governing permissions and limitations under the License. + +#include "common/FeatureReport.h" + +#include + +#include + +#include "common/PrometheusClient.h" + +namespace milvus::monitor::feature_report { +namespace { + +constexpr std::string_view kSourceCpp = "cpp"; +constexpr auto kReportInterval = std::chrono::hours(1); + +prometheus::Family& +FeatureReportFamily() { + static auto& family = prometheus::BuildCounter() + .Name("milvus_feature_report_total") + .Help("Count of feature reports emitted.") + .Register(milvus::monitor::getPrometheusClient().GetRegistry()); + return family; +} + +prometheus::Counter& +CounterForFeature(std::string_view feature) { + return FeatureReportFamily().Add({{"feature", std::string(feature)}, {"source", std::string(kSourceCpp)}}); +} + +int64_t +ToNanos(std::chrono::steady_clock::time_point time_point) { + return std::chrono::duration_cast(time_point.time_since_epoch()).count(); +} + +} // namespace + +FeatureReporter::FeatureReporter(std::string_view name) : name_(name), counter_(CounterForFeature(name_)) { +} + +std::string_view +FeatureReporter::Name() const noexcept { + return std::string_view(name_); +} + +bool +FeatureReporter::Record() { + return recordAt(std::chrono::steady_clock::now()); +} + +void +FeatureReporter::reset() { + next_allowed_nanos_.store(0, std::memory_order_release); +} + +bool +FeatureReporter::recordAt(std::chrono::steady_clock::time_point now) { + if (name_.empty()) { + return false; + } + + const auto now_nanos = ToNanos(now); + const auto next_nanos = ToNanos(now + kReportInterval); + + auto old = next_allowed_nanos_.load(std::memory_order_acquire); + while (now_nanos >= old) { + if (next_allowed_nanos_.compare_exchange_weak(old, next_nanos, std::memory_order_acq_rel, + std::memory_order_acquire)) { + counter_.Increment(); + return true; + } + } + return false; +} + +#define DEFINE_FEATURE_REPORTER(name, label) \ + FeatureReporter& name() { \ + static FeatureReporter reporter{k##name}; \ + return reporter; \ + } + +MILVUS_FEATURE_REPORTERS(DEFINE_FEATURE_REPORTER) + +#undef DEFINE_FEATURE_REPORTER + +} // namespace milvus::monitor::feature_report diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index d23b993..bab66a8 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -16,6 +16,7 @@ find_package(Threads REQUIRED) set(ALL_TEST_FILES init_gtest.cpp + FeatureReportTest.cpp TracerTest.cpp StreamTest.cpp ) diff --git a/test/FeatureReportTest.cpp b/test/FeatureReportTest.cpp new file mode 100644 index 0000000..fff2072 --- /dev/null +++ b/test/FeatureReportTest.cpp @@ -0,0 +1,111 @@ +// Copyright (C) 2019-2020 Zilliz. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software distributed under the License +// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +// or implied. See the License for the specific language governing permissions and limitations under the License. + +#include + +#include +#include +#include +#include +#include +#include + +#include "common/FeatureReport.h" +#include "common/PrometheusClient.h" + +namespace milvus::monitor::feature_report { + +struct FeatureReporterTestPeer { + static bool + RecordAt(FeatureReporter& reporter, std::chrono::steady_clock::time_point now) { + return reporter.recordAt(now); + } + + static void + Reset(FeatureReporter& reporter) { + reporter.reset(); + } +}; + +namespace { + +double +FeatureReportValue(std::string_view feature) { + for (const auto& family : getPrometheusClient().GetRegistry().Collect()) { + if (family.name != "milvus_feature_report_total" || family.type != prometheus::MetricType::Counter) { + continue; + } + + for (const auto& metric : family.metric) { + bool feature_match = false; + bool source_match = false; + for (const auto& label : metric.label) { + if (label.name == "feature" && std::string_view(label.value) == feature) { + feature_match = true; + } else if (label.name == "source" && label.value == "cpp") { + source_match = true; + } + } + if (feature_match && source_match) { + return metric.counter.value; + } + } + } + return 0.0; +} + +TEST(FeatureReportTest, Throttle) { + auto& reporter = HybridSearch(); + FeatureReporterTestPeer::Reset(reporter); + const auto before = FeatureReportValue(reporter.Name()); + const auto now = std::chrono::steady_clock::time_point(std::chrono::seconds(100)); + + ASSERT_TRUE(FeatureReporterTestPeer::RecordAt(reporter, now)); + ASSERT_FALSE(FeatureReporterTestPeer::RecordAt(reporter, now + std::chrono::minutes(1))); + ASSERT_TRUE(FeatureReporterTestPeer::RecordAt(reporter, now + std::chrono::hours(1))); + + ASSERT_DOUBLE_EQ(FeatureReportValue(reporter.Name()) - before, 2.0); +} + +TEST(FeatureReportTest, ConcurrentCalls) { + auto& reporter = PartitionKey(); + FeatureReporterTestPeer::Reset(reporter); + const auto before = FeatureReportValue(reporter.Name()); + const auto now = std::chrono::steady_clock::time_point(std::chrono::seconds(200)); + constexpr int kThreads = 32; + + std::atomic reported{0}; + std::vector threads; + threads.reserve(kThreads); + for (int i = 0; i < kThreads; ++i) { + threads.emplace_back([&] { + if (FeatureReporterTestPeer::RecordAt(reporter, now)) { + reported.fetch_add(1, std::memory_order_relaxed); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + + ASSERT_EQ(reported.load(std::memory_order_relaxed), 1); + ASSERT_DOUBLE_EQ(FeatureReportValue(reporter.Name()) - before, 1.0); +} + +TEST(FeatureReportTest, PredeclaredReporter) { + FeatureReporterTestPeer::Reset(DynamicField()); + ASSERT_EQ(DynamicField().Name(), kDynamicField); + ASSERT_TRUE(FeatureReporterTestPeer::RecordAt(DynamicField(), + std::chrono::steady_clock::time_point(std::chrono::seconds(300)))); +} + +} // namespace +} // namespace milvus::monitor::feature_report