services:
- rgw
with_legacy: true
+
+- name: rgw_bucket_persistent_notif_num_shards
+ type: uint
+ level: advanced
+ desc: Number of shards for a persistent topic.
+ long_desc: Number of shards of persistent topics. The notifications will be sharded by a combination of
+ the bucket and key name. Changing the number effect only new topics and does not change exiting ones.
+ default: 11
+ services:
+ - rgw
+
+- name: rgw_usage_stats_refresh_interval
+ type: int
+ level: advanced
+ desc: Interval in seconds for background refresh of usage statistics for active buckets/users
+ default: 5_min
+ services:
+ - rgw
+ see_also:
+ - rgw_user_quota_sync_interval
+ with_legacy: true
ldpp_dout(dpp, 10) << "Usage performance counters initialized successfully" << dendl;
}
}
+
+ if (!cache_config.db_path.empty()) {
+
+ usage_perf_counters = std::make_unique<rgw::UsagePerfCounters>(
+ dpp->get_cct(),
+ env.driver, // the driver parameter
+ cache_config);
+
+ int r = usage_perf_counters->init();
+ if (r < 0) {
+ ldpp_dout(dpp, 1) << "WARNING: Failed to initialize usage perf counters: "
+ << cpp_strerror(-r) << " (continuing without them)" << dendl;
+ usage_perf_counters.reset();
+ } else {
+ rgw::set_usage_perf_counters(usage_perf_counters.get());
+ ldpp_dout(dpp, 10) << "Usage performance counters initialized successfully" << dendl;
+ }
+ }
}
} /* init_perfcounters */
}
void RGWOp::update_usage_stats_if_needed() {
- auto* usage_counters = rgw::get_usage_perf_counters();
- if (!usage_counters) {
- return;
- }
-
- // Only update for successful operations
- if (op_ret != 0) {
- return;
- }
-
- // Check if this is an operation that changes usage
- bool is_put_op = (dynamic_cast<RGWPutObj*>(this) != nullptr) ||
- (dynamic_cast<RGWPostObj*>(this) != nullptr) ||
- (dynamic_cast<RGWCopyObj*>(this) != nullptr) ||
- (dynamic_cast<RGWCompleteMultipart*>(this) != nullptr);
-
- bool is_delete_op = (dynamic_cast<RGWDeleteObj*>(this) != nullptr) ||
- (dynamic_cast<RGWDeleteMultiObj*>(this) != nullptr);
-
- if (!is_put_op && !is_delete_op) {
- return;
- }
-
- // Update bucket statistics if we have bucket info
- if (s->bucket && usage_counters) {
- try {
- // Use sync_owner_stats to get current bucket stats
- RGWBucketEnt ent;
- int ret = s->bucket->sync_owner_stats(this, null_yield, &ent);
-
- if (ret >= 0) {
- // Update bucket stats with accurate counts
- usage_counters->update_bucket_stats(
- s->bucket->get_name(),
- ent.size, // Total bytes used
- ent.count // Total number of objects
- );
-
- ldout(s->cct, 20) << "Updated bucket stats for " << s->bucket->get_name()
- << ": bytes=" << ent.size
- << ", objects=" << ent.count << dendl;
- } else {
- ldout(s->cct, 10) << "Failed to sync bucket stats for "
- << s->bucket->get_name()
- << ": " << cpp_strerror(-ret) << dendl;
- }
- } catch (const std::exception& e) {
- ldout(s->cct, 5) << "Exception updating bucket stats: " << e.what() << dendl;
- }
- }
-
- // Update user statistics
- if (s->user && usage_counters) {
- try {
- // Initialize user stats
- RGWStorageStats user_stats;
- user_stats.size_utilized = 0;
- user_stats.num_objects = 0;
-
- // Try to list buckets and sum up stats
- rgw::sal::BucketList buckets;
- rgw_owner owner(s->user->get_id());
-
- // list_buckets signature: (dpp, owner, marker, end_marker, max, need_stats, buckets, y)
- int ret = driver->list_buckets(this, owner, "", "", "", 1000, true, buckets, null_yield);
-
- if (ret >= 0) {
- // Sum up stats from all buckets
- // The buckets.buckets container holds RGWBucketEnt objects directly
- for (const auto& bucket_ent : buckets.buckets) {
- user_stats.size_utilized += bucket_ent.size;
- user_stats.num_objects += bucket_ent.count;
- }
-
- // Update user stats with the total
- usage_counters->update_user_stats(
- s->user->get_id().id,
- user_stats.size_utilized,
- user_stats.num_objects
- );
-
- ldout(s->cct, 20) << "Updated user stats for " << s->user->get_id().id
- << " (from " << buckets.buckets.size() << " buckets)"
- << ": bytes=" << user_stats.size_utilized
- << ", objects=" << user_stats.num_objects << dendl;
- } else {
- // Fallback: Use current bucket stats as approximation
- if (s->bucket) {
- RGWBucketEnt ent;
- ret = s->bucket->sync_owner_stats(this, null_yield, &ent);
-
- if (ret >= 0) {
- // Get existing cached user stats
- auto cached_stats = usage_counters->get_user_stats(s->user->get_id().id);
-
- uint64_t total_bytes = ent.size;
- uint64_t total_objects = ent.count;
-
- // If we have cached stats and this is a different bucket, try to accumulate
- if (cached_stats.has_value()) {
- // Use the maximum of cached and current values
- // This prevents losing data when we can't list all buckets
- total_bytes = std::max(cached_stats->bytes_used, ent.size);
- total_objects = std::max(cached_stats->num_objects, ent.count);
- }
-
- usage_counters->update_user_stats(
- s->user->get_id().id,
- total_bytes,
- total_objects
- );
-
- ldout(s->cct, 20) << "Updated user stats (partial) for " << s->user->get_id().id
- << ": bytes=" << total_bytes
- << ", objects=" << total_objects << dendl;
- }
- }
- }
- } catch (const std::exception& e) {
- ldout(s->cct, 5) << "Exception updating user stats: " << e.what() << dendl;
- }
- }
+ // Stats are now updated directly in each operation's execute() method
+ // This method can be left as a no-op.
}
int RGWGetObj::handle_slo_manifest(bufferlist& bl, optional_yield y)
ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
// too late to rollback operation, hence op_ret is not set here
}
+
+ // Update usage statistics after successful upload
+ if (op_ret == 0 && s->bucket && s->obj_size > 0) {
+ auto usage_counters = rgw::get_usage_perf_counters();
+ if (usage_counters) {
+ std::string bucket_key = s->bucket->get_tenant().empty() ?
+ s->bucket->get_name() :
+ s->bucket->get_tenant() + "/" + s->bucket->get_name();
+
+ ldpp_dout(this, 20) << "PUT completed: updating usage for bucket="
+ << bucket_key << " size=" << s->obj_size << dendl;
+
+ // Increment bucket stats in cache (adds to existing)
+ auto existing = usage_counters->get_bucket_stats(bucket_key);
+ uint64_t new_bytes = s->obj_size;
+ uint64_t new_objects = 1;
+
+ if (existing) {
+ new_bytes += existing->bytes_used;
+ new_objects += existing->num_objects;
+ }
+
+ // Update cache with new totals
+ usage_counters->update_bucket_stats(s->bucket->get_name(),
+ new_bytes, new_objects, true);
+
+ // Mark as active to trigger perf counter update
+ usage_counters->mark_bucket_active(s->bucket->get_name(),
+ s->bucket->get_tenant());
+
+ // Update user stats
+ if (s->user) {
+ auto user_existing = usage_counters->get_user_stats(s->user->get_id().to_str());
+ uint64_t user_bytes = s->obj_size;
+ uint64_t user_objects = 1;
+
+ if (user_existing) {
+ user_bytes += user_existing->bytes_used;
+ user_objects += user_existing->num_objects;
+ }
+
+ usage_counters->update_user_stats(s->user->get_id().to_str(),
+ user_bytes, user_objects, true);
+ usage_counters->mark_user_active(s->user->get_id().to_str());
+ }
+ }
+ }
+
} /* RGWPutObj::execute() */
int RGWPostObj::init_processing(optional_yield y)
} else {
op_ret = -EINVAL;
}
+
+ // Update usage statistics after successful delete
+ if (op_ret == 0 && s->bucket) {
+ auto usage_counters = rgw::get_usage_perf_counters();
+ if (usage_counters) {
+ std::string bucket_key = s->bucket->get_tenant().empty() ?
+ s->bucket->get_name() :
+ s->bucket->get_tenant() + "/" + s->bucket->get_name();
+
+ ldpp_dout(this, 20) << "DELETE completed: updating usage for bucket="
+ << bucket_key << dendl;
+
+ // Decrement bucket stats in cache
+ auto existing = usage_counters->get_bucket_stats(bucket_key);
+ if (existing && existing->num_objects > 0) {
+ uint64_t obj_size = existing->bytes_used / existing->num_objects; // approximate
+ uint64_t new_bytes = existing->bytes_used > obj_size ?
+ existing->bytes_used - obj_size : 0;
+ uint64_t new_objects = existing->num_objects - 1;
+
+ usage_counters->update_bucket_stats(s->bucket->get_name(),
+ new_bytes, new_objects, true);
+ }
+
+ // Mark as active to trigger perf counter update
+ usage_counters->mark_bucket_active(s->bucket->get_name(),
+ s->bucket->get_tenant());
+
+ if (s->user) {
+ usage_counters->mark_user_active(s->user->get_id().to_str());
+ }
+ }
+ }
+
}
bool RGWCopyObj::parse_copy_location(const std::string_view& url_src,
int clear_expired_entries();
size_t get_cache_size() const;
+ const Config& get_config() const { return config; }
+
// Performance metrics
uint64_t get_cache_hits() const;
uint64_t get_cache_misses() const;
#include "common/dout.h"
#include "common/perf_counters_collection.h"
#include "common/errno.h"
+#include "rgw_sal.h"
+#include "rgw_sal_rados.h"
+#include "rgw_bucket.h"
+#include "rgw_user.h"
+#include "common/async/yield_context.h"
#define dout_subsys ceph_subsys_rgw
UsagePerfCounters::UsagePerfCounters(CephContext* cct,
const UsageCache::Config& cache_config)
- : cct(cct), cache(std::make_unique<UsageCache>(cct, cache_config)),
- global_counters(nullptr) {
+ : cct(cct), driver(nullptr),
+ cache(std::make_unique<UsageCache>(cct, cache_config)),
+ global_counters(nullptr)
+{
+ create_global_counters();
+}
+
+UsagePerfCounters::UsagePerfCounters(CephContext* cct)
+ : UsagePerfCounters(cct, UsageCache::Config{}) {}
+
+UsagePerfCounters::UsagePerfCounters(CephContext* cct,
+ rgw::sal::Driver* driver,
+ const UsageCache::Config& cache_config)
+ : cct(cct),
+ driver(driver), // ADD THIS
+ cache(std::make_unique<UsageCache>(cct, cache_config)),
+ global_counters(nullptr)
+{
create_global_counters();
}
+// Update the destructor
UsagePerfCounters::~UsagePerfCounters() {
shutdown();
}
return ret;
}
+ create_global_counters();
ldout(cct, 10) << "Usage performance counters initialized successfully" << dendl;
return 0;
}
void UsagePerfCounters::start() {
ldout(cct, 10) << "Starting usage perf counters" << dendl;
+ shutdown_flag = false;
// Start cleanup thread
cleanup_thread = std::thread(&UsagePerfCounters::cleanup_worker, this);
+
+ // Start refresh thread
+ refresh_thread = std::thread(&UsagePerfCounters::refresh_worker, this);
+
+ ldout(cct, 10) << "Started usage perf counters threads" << dendl;
}
void UsagePerfCounters::stop() {
ldout(cct, 10) << "Stopping usage perf counters" << dendl;
- // Stop cleanup thread
+ // Signal threads to stop
shutdown_flag = true;
+
+ // Wait for cleanup thread
if (cleanup_thread.joinable()) {
cleanup_thread.join();
}
+
+ // Wait for refresh thread
+ if (refresh_thread.joinable()) {
+ refresh_thread.join();
+ }
+
+ ldout(cct, 10) << "Stopped usage perf counters threads" << dendl;
}
void UsagePerfCounters::shutdown() {
- stop();
+ shutdown_flag = true;
+
+ if (cleanup_thread.joinable()) {
+ cleanup_thread.join();
+ }
+
+ if (refresh_thread.joinable()) {
+ refresh_thread.join();
+ }
// Clean up perf counters
{
// Shutdown cache
cache->shutdown();
- ldout(cct, 10) << "Usage perf counters shutdown complete" << dendl;
+ ldout(cct, 10) << "Shutdown usage perf counters" << dendl;
+}
+void UsagePerfCounters::update_bucket_stats(const std::string& bucket_name,
+ uint64_t bytes_used,
+ uint64_t num_objects,
+ bool update_cache) {
+ ldout(cct, 20) << "update_bucket_stats: bucket=" << bucket_name
+ << " bytes=" << bytes_used
+ << " objects=" << num_objects
+ << " update_cache=" << update_cache << dendl;
+
+ // Update cache if requested - ALWAYS write the total values passed in
+ if (update_cache && cache) {
+ int ret = cache->update_bucket_stats(bucket_name, bytes_used, num_objects);
+ if (ret == 0) {
+ global_counters->inc(l_rgw_usage_cache_update);
+ ldout(cct, 15) << "Cache updated for bucket " << bucket_name
+ << " total_bytes=" << bytes_used
+ << " total_objects=" << num_objects << dendl;
+ } else {
+ ldout(cct, 5) << "Failed to update bucket cache: " << cpp_strerror(-ret) << dendl;
+ }
+ }
+
+ // Define local enum
+ enum {
+ l_rgw_bucket_first = 940000,
+ l_rgw_bucket_bytes,
+ l_rgw_bucket_objects,
+ l_rgw_bucket_last
+ };
+
+ // Update perf counters
+ {
+ std::unique_lock lock(counters_mutex);
+
+ auto it = bucket_perf_counters.find(bucket_name);
+ if (it == bucket_perf_counters.end()) {
+ PerfCounters* counters = create_bucket_counters(bucket_name);
+ if (counters) {
+ bucket_perf_counters[bucket_name] = counters;
+ it = bucket_perf_counters.find(bucket_name);
+ ldout(cct, 15) << "Created perf counter for bucket " << bucket_name << dendl;
+ }
+ }
+
+ if (it != bucket_perf_counters.end() && it->second) {
+ it->second->set(l_rgw_bucket_bytes, bytes_used);
+ it->second->set(l_rgw_bucket_objects, num_objects);
+ ldout(cct, 15) << "Set perf counter for bucket " << bucket_name
+ << " bytes=" << bytes_used
+ << " objects=" << num_objects << dendl;
+ }
+ }
}
void UsagePerfCounters::update_user_stats(const std::string& user_id,
uint64_t bytes_used,
uint64_t num_objects,
bool update_cache) {
- // Update cache if requested
+ ldout(cct, 20) << "update_user_stats: user=" << user_id
+ << " bytes=" << bytes_used
+ << " objects=" << num_objects
+ << " update_cache=" << update_cache << dendl;
+
+ // Update cache if requested - ALWAYS write the total values passed in
if (update_cache && cache) {
int ret = cache->update_user_stats(user_id, bytes_used, num_objects);
if (ret == 0) {
global_counters->inc(l_rgw_usage_cache_update);
+ ldout(cct, 15) << "Cache updated for user " << user_id
+ << " total_bytes=" << bytes_used
+ << " total_objects=" << num_objects << dendl;
} else {
- ldout(cct, 5) << "Failed to update user cache for " << user_id
- << ": " << cpp_strerror(-ret) << dendl;
+ ldout(cct, 5) << "Failed to update user cache: " << cpp_strerror(-ret) << dendl;
}
}
- // Define local enum for user-specific counter indices
- // This avoids needing placeholders in the global enum
+ // Define local enum
enum {
- l_rgw_user_first = 930000, // Start at a high number to avoid conflicts
- l_rgw_user_bytes, // 930001
- l_rgw_user_objects, // 930002
- l_rgw_user_last // 930003
+ l_rgw_user_first = 930000,
+ l_rgw_user_bytes,
+ l_rgw_user_objects,
+ l_rgw_user_last
};
- // Update or create perf counters
+ // Update perf counters
{
std::unique_lock lock(counters_mutex);
auto it = user_perf_counters.find(user_id);
if (it == user_perf_counters.end()) {
- // Counter doesn't exist, create it
PerfCounters* counters = create_user_counters(user_id);
- user_perf_counters[user_id] = counters;
- it = user_perf_counters.find(user_id);
+ if (counters) {
+ user_perf_counters[user_id] = counters;
+ it = user_perf_counters.find(user_id);
+ ldout(cct, 15) << "Created perf counter for user " << user_id << dendl;
+ }
}
- // Set the values using the local enum indices
- it->second->set(l_rgw_user_bytes, bytes_used);
- it->second->set(l_rgw_user_objects, num_objects);
+ if (it != user_perf_counters.end() && it->second) {
+ it->second->set(l_rgw_user_bytes, bytes_used);
+ it->second->set(l_rgw_user_objects, num_objects);
+ ldout(cct, 15) << "Set perf counter for user " << user_id
+ << " bytes=" << bytes_used
+ << " objects=" << num_objects << dendl;
+ }
}
-
- ldout(cct, 20) << "Updated user stats: " << user_id
- << " bytes=" << bytes_used
- << " objects=" << num_objects << dendl;
}
-void UsagePerfCounters::update_bucket_stats(const std::string& bucket_name,
- uint64_t bytes_used,
- uint64_t num_objects,
- bool update_cache) {
- // Update cache if requested
- if (update_cache && cache) {
- int ret = cache->update_bucket_stats(bucket_name, bytes_used, num_objects);
- if (ret == 0) {
- global_counters->inc(l_rgw_usage_cache_update);
- } else {
- ldout(cct, 5) << "Failed to update bucket cache for " << bucket_name
- << ": " << cpp_strerror(-ret) << dendl;
+void UsagePerfCounters::mark_bucket_active(const std::string& bucket_name,
+ const std::string& tenant) {
+ std::string key = tenant.empty() ? bucket_name : tenant + "/" + bucket_name;
+
+ ldout(cct, 20) << "mark_bucket_active: key=" << key << dendl;
+
+ // Add to active set for background refresh
+ {
+ std::lock_guard<std::mutex> lock(activity_mutex);
+ active_buckets.insert(key);
+ }
+
+ // Ensure perf counter exists
+ {
+ std::unique_lock lock(counters_mutex);
+ if (bucket_perf_counters.find(key) == bucket_perf_counters.end()) {
+ PerfCounters* pc = create_bucket_counters(key);
+ if (pc) {
+ bucket_perf_counters[key] = pc;
+ }
}
}
- // Define local enum for bucket-specific counter indices
- // This avoids needing placeholders in the global enum
- enum {
- l_rgw_bucket_first = 940000, // Different range from user counters
- l_rgw_bucket_bytes, // 940001
- l_rgw_bucket_objects, // 940002
- l_rgw_bucket_last // 940003
- };
+ // Immediately update from cache
+ auto cached_stats = cache->get_bucket_stats(key);
+ if (cached_stats) {
+ ldout(cct, 15) << "Updating from cache: bucket=" << key
+ << " bytes=" << cached_stats->bytes_used
+ << " objects=" << cached_stats->num_objects << dendl;
+
+ std::string bucket_only = key;
+ size_t pos = key.find('/');
+ if (pos != std::string::npos) {
+ bucket_only = key.substr(pos + 1);
+ }
+
+ update_bucket_stats(bucket_only, cached_stats->bytes_used,
+ cached_stats->num_objects, false);
+ }
+}
+
+void UsagePerfCounters::mark_user_active(const std::string& user_id) {
+ ldout(cct, 20) << "mark_user_active: user=" << user_id << dendl;
- // Update or create perf counters
+ // Add to active set for background refresh
+ {
+ std::lock_guard<std::mutex> lock(activity_mutex);
+ active_users.insert(user_id);
+ }
+
+ // Ensure perf counter exists
{
std::unique_lock lock(counters_mutex);
+ if (user_perf_counters.find(user_id) == user_perf_counters.end()) {
+ PerfCounters* pc = create_user_counters(user_id);
+ if (pc) {
+ user_perf_counters[user_id] = pc;
+ }
+ }
+ }
+
+ // Immediately update from cache
+ auto cached_stats = cache->get_user_stats(user_id);
+ if (cached_stats) {
+ ldout(cct, 15) << "Updating from cache: user=" << user_id
+ << " bytes=" << cached_stats->bytes_used
+ << " objects=" << cached_stats->num_objects << dendl;
- auto it = bucket_perf_counters.find(bucket_name);
- if (it == bucket_perf_counters.end()) {
- // Counter doesn't exist, create it
- PerfCounters* counters = create_bucket_counters(bucket_name);
- bucket_perf_counters[bucket_name] = counters;
- it = bucket_perf_counters.find(bucket_name);
+ update_user_stats(user_id, cached_stats->bytes_used,
+ cached_stats->num_objects, false);
+ }
+}
+
+void UsagePerfCounters::refresh_worker() {
+ ldout(cct, 10) << "Started usage stats refresh worker thread" << dendl;
+
+ while (!shutdown_flag) {
+ // Sleep for the refresh interval with periodic checks for shutdown
+ auto sleep_until = std::chrono::steady_clock::now() + refresh_interval;
+
+ while (!shutdown_flag && std::chrono::steady_clock::now() < sleep_until) {
+ std::this_thread::sleep_for(std::chrono::seconds(1));
}
- // Set the values using the local enum indices
- it->second->set(l_rgw_bucket_bytes, bytes_used);
- it->second->set(l_rgw_bucket_objects, num_objects);
+ if (shutdown_flag) {
+ break;
+ }
+
+ // Get snapshot of active buckets and users
+ std::unordered_set<std::string> buckets_to_refresh;
+ std::unordered_set<std::string> users_to_refresh;
+
+ {
+ std::lock_guard<std::mutex> lock(activity_mutex);
+ buckets_to_refresh = active_buckets;
+ users_to_refresh = active_users;
+ }
+
+ ldout(cct, 15) << "Background refresh: checking " << buckets_to_refresh.size()
+ << " buckets and " << users_to_refresh.size()
+ << " users" << dendl;
+
+ // Refresh bucket stats from cache
+ for (const auto& bucket_key : buckets_to_refresh) {
+ if (shutdown_flag) break;
+ refresh_bucket_stats(bucket_key);
+ }
+
+ // Refresh user stats from cache
+ for (const auto& user_id : users_to_refresh) {
+ if (shutdown_flag) break;
+ refresh_user_stats(user_id);
+ }
+ }
+
+ ldout(cct, 10) << "Stopped usage stats refresh worker thread" << dendl;
+}
+
+void UsagePerfCounters::refresh_bucket_stats(const std::string& bucket_key) {
+ ldout(cct, 20) << "refresh_bucket_stats: key=" << bucket_key << dendl;
+
+ auto cached_stats = cache->get_bucket_stats(bucket_key);
+ if (cached_stats) {
+ std::string bucket_name = bucket_key;
+ size_t pos = bucket_key.find('/');
+ if (pos != std::string::npos) {
+ bucket_name = bucket_key.substr(pos + 1);
+ }
+
+ ldout(cct, 15) << "Refreshing bucket " << bucket_key
+ << " bytes=" << cached_stats->bytes_used
+ << " objects=" << cached_stats->num_objects << dendl;
+
+ update_bucket_stats(bucket_name, cached_stats->bytes_used,
+ cached_stats->num_objects, false);
}
+}
+
+void UsagePerfCounters::refresh_user_stats(const std::string& user_id) {
+ ldout(cct, 20) << "refresh_user_stats: user=" << user_id << dendl;
- ldout(cct, 20) << "Updated bucket stats: " << bucket_name
- << " bytes=" << bytes_used
- << " objects=" << num_objects << dendl;
+ auto cached_stats = cache->get_user_stats(user_id);
+ if (cached_stats) {
+ ldout(cct, 15) << "Refreshing user " << user_id
+ << " bytes=" << cached_stats->bytes_used
+ << " objects=" << cached_stats->num_objects << dendl;
+
+ update_user_stats(user_id, cached_stats->bytes_used,
+ cached_stats->num_objects, false);
+ }
}
void UsagePerfCounters::refresh_from_cache(const std::string& user_id,
#include "common/perf_counters.h"
#include "rgw_usage_cache.h"
+// To avoid heavy header rgw_sal.h , we are adding a forward declaration here
+namespace rgw::sal {
+ class Driver;
+ class Bucket;
+ class User;
+}
namespace rgw {
class UsagePerfCounters {
private:
CephContext* cct;
+ rgw::sal::Driver* driver;
std::unique_ptr<UsageCache> cache;
mutable std::shared_mutex counters_mutex;
PerfCounters* global_counters;
+ // Track active buckets and users for background refresh
+ std::unordered_set<std::string> active_buckets;
+ std::unordered_set<std::string> active_users;
+ mutable std::mutex activity_mutex;
+
// Cleanup thread management
std::thread cleanup_thread;
+ std::thread refresh_thread;
std::atomic<bool> shutdown_flag{false};
std::chrono::seconds cleanup_interval{300}; // 5 minutes
+ std::chrono::seconds refresh_interval{60};
void create_global_counters();
PerfCounters* create_user_counters(const std::string& user_id);
PerfCounters* create_bucket_counters(const std::string& bucket_name);
void cleanup_worker();
+ void refresh_worker();
+
+ void refresh_bucket_stats(const std::string& bucket_key);
+ void refresh_user_stats(const std::string& user_id);
public:
explicit UsagePerfCounters(CephContext* cct,
const UsageCache::Config& cache_config);
- explicit UsagePerfCounters(CephContext* cct)
- : UsagePerfCounters(cct, UsageCache::Config{}) {}
+ UsagePerfCounters(CephContext* cct);
+ UsagePerfCounters(CephContext* cct,
+ rgw::sal::Driver* driver,
+ const UsageCache::Config& cache_config);
~UsagePerfCounters();
// Lifecycle management
uint64_t bytes_used,
uint64_t num_objects,
bool update_cache = true);
-
+
+void mark_bucket_active(const std::string& bucket_name,
+ const std::string& tenant = "");
+
+void mark_user_active(const std::string& user_id);
+
// Cache operations
void refresh_from_cache(const std::string& user_id,
const std::string& bucket_name);
void set_cleanup_interval(std::chrono::seconds interval) {
cleanup_interval = interval;
}
+
+ void set_refresh_interval(std::chrono::seconds interval) {
+ refresh_interval = interval;
+ }
};
// Global singleton access
${CMAKE_SOURCE_DIR}/src/rgw/rgw_usage_cache.cc
${CMAKE_SOURCE_DIR}/src/rgw/rgw_usage_perf.cc)
+target_include_directories(unittest_rgw_usage_perf_counters PRIVATE
+ ${CMAKE_SOURCE_DIR}/src/rgw
+ ${CMAKE_SOURCE_DIR}/src/rgw/services
+ ${CMAKE_SOURCE_DIR}/src/rgw/driver/rados
+ ${CMAKE_SOURCE_DIR}/src/rgw/store/rados
+ ${CMAKE_SOURCE_DIR}/src/dmclock/support/src
+ ${CMAKE_SOURCE_DIR}/src/dmclock/src
+)
+
target_link_libraries(unittest_rgw_usage_perf_counters
${UNITTEST_LIBS}
global
<< num_buckets << " buckets" << std::endl;
}
+TEST_F(TestRGWUsageCache, ManualTestScenario_UserAndBuckets) {
+ // Simulate the complete manual test scenario:
+ // User "testuser" with bucket1 (2 files) and bucket2 (1 file)
+
+ std::cout << "\n=== Manual Test Scenario: User and Buckets ===" << std::endl;
+
+ const std::string user_id = "testuser";
+ const std::string bucket1 = "bucket1";
+ const std::string bucket2 = "bucket2";
+
+ // File sizes from manual test
+ const uint64_t small_txt = 12; // "Hello World\n"
+ const uint64_t medium_bin = 5 * 1024 * 1024; // 5MB
+ const uint64_t large_bin = 10 * 1024 * 1024; // 10MB
+
+ // Bucket1: small.txt + medium.bin
+ const uint64_t bucket1_bytes = small_txt + medium_bin;
+ const uint64_t bucket1_objects = 2;
+
+ // Bucket2: large.bin
+ const uint64_t bucket2_bytes = large_bin;
+ const uint64_t bucket2_objects = 1;
+
+ // User totals
+ const uint64_t user_bytes = bucket1_bytes + bucket2_bytes;
+ const uint64_t user_objects = 3;
+
+ std::cout << "Uploading files to buckets:" << std::endl;
+ std::cout << " " << bucket1 << ": small.txt (12B) + medium.bin (5MB) = "
+ << bucket1_bytes << " bytes, " << bucket1_objects << " objects" << std::endl;
+ std::cout << " " << bucket2 << ": large.bin (10MB) = "
+ << bucket2_bytes << " bytes, " << bucket2_objects << " objects" << std::endl;
+
+ // Update cache
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket1, bucket1_bytes, bucket1_objects));
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket2, bucket2_bytes, bucket2_objects));
+ ASSERT_EQ(0, cache->update_user_stats(user_id, user_bytes, user_objects));
+
+ // Verify bucket1
+ auto b1_stats = cache->get_bucket_stats(bucket1);
+ ASSERT_TRUE(b1_stats.has_value());
+ EXPECT_EQ(bucket1_bytes, b1_stats->bytes_used);
+ EXPECT_EQ(bucket1_objects, b1_stats->num_objects);
+ std::cout << " Bucket1 verified: " << b1_stats->bytes_used << " bytes, "
+ << b1_stats->num_objects << " objects" << std::endl;
+
+ // Verify bucket2
+ auto b2_stats = cache->get_bucket_stats(bucket2);
+ ASSERT_TRUE(b2_stats.has_value());
+ EXPECT_EQ(bucket2_bytes, b2_stats->bytes_used);
+ EXPECT_EQ(bucket2_objects, b2_stats->num_objects);
+ std::cout << " Bucket2 verified: " << b2_stats->bytes_used << " bytes, "
+ << b2_stats->num_objects << " objects" << std::endl;
+
+ // Verify user totals
+ auto user_stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(user_stats.has_value());
+ EXPECT_EQ(user_bytes, user_stats->bytes_used);
+ EXPECT_EQ(user_objects, user_stats->num_objects);
+
+ std::cout << " User verified: " << user_stats->bytes_used << " bytes (~"
+ << (user_stats->bytes_used / (1024 * 1024)) << "MB), "
+ << user_stats->num_objects << " objects" << std::endl;
+
+ // Verify expected totals
+ EXPECT_GE(user_bytes, 15 * 1024 * 1024) << "User should have ~15MB";
+ EXPECT_EQ(3u, user_objects) << "User should have exactly 3 objects";
+
+ std::cout << " All statistics match manual test expectations" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, RepeatedAccessCacheHits) {
+ // Simulate: s3cmd ls s3://bucket1 (multiple times)
+ // Validate cache hit behavior
+
+ std::cout << "\n=== Testing Repeated Access Cache Hits ===" << std::endl;
+
+ const std::string bucket_name = "bucket1";
+
+ // Add bucket to cache
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 5 * 1024 * 1024, 2));
+ std::cout << "Added bucket to cache" << std::endl;
+
+ // Simulate multiple s3cmd ls operations
+ std::cout << "Simulating multiple bucket listings:" << std::endl;
+ for (int i = 0; i < 3; ++i) {
+ auto stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value()) << "Failed on iteration " << (i + 1);
+ std::cout << " Listing " << (i + 1) << ": Found bucket with "
+ << stats->bytes_used << " bytes" << std::endl;
+ }
+
+ std::cout << " All listings successful - bucket data retrieved from cache" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, PerformanceNoSyncOwnerStats) {
+ // CRITICAL: Verify cache operations are fast (not 90ms like cls_bucket_head)
+
+ std::cout << "\n=== Performance Test: No sync_owner_stats Blocking ===" << std::endl;
+
+ const int num_operations = 1000;
+ std::vector<std::string> bucket_names;
+
+ // Pre-populate cache with buckets
+ for (int i = 0; i < 10; ++i) {
+ std::string name = "perf_bucket_" + std::to_string(i);
+ bucket_names.push_back(name);
+ ASSERT_EQ(0, cache->update_bucket_stats(name, i * 1024, i));
+ }
+
+ std::cout << "Running " << num_operations << " cache operations..." << std::endl;
+
+ // Measure time for many get operations
+ auto start = std::chrono::high_resolution_clock::now();
+
+ for (int i = 0; i < num_operations; ++i) {
+ std::string& bucket_name = bucket_names[i % bucket_names.size()];
+ auto stats = cache->get_bucket_stats(bucket_name);
+ // All should be cache hits
+ }
+
+ auto end = std::chrono::high_resolution_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
+
+ double total_ms = duration.count() / 1000.0;
+ double avg_us = static_cast<double>(duration.count()) / num_operations;
+ double avg_ms = avg_us / 1000.0;
+
+ std::cout << "\nPerformance Results:" << std::endl;
+ std::cout << " Total operations: " << num_operations << std::endl;
+ std::cout << " Total time: " << total_ms << " ms" << std::endl;
+ std::cout << " Average per operation: " << avg_us << " μs (" << avg_ms << " ms)" << std::endl;
+
+ // CRITICAL: Should be WAY faster than 90ms
+ EXPECT_LT(avg_ms, 10.0)
+ << "Operations too slow (" << avg_ms << "ms) - possible sync_owner_stats in path";
+
+ // Should be sub-millisecond for in-memory cache
+ EXPECT_LT(avg_ms, 1.0)
+ << "Cache operations should be < 1ms, got " << avg_ms << "ms";
+
+ double speedup = 90.0 / avg_ms;
+ std::cout << " Performance test PASSED!" << std::endl;
+ std::cout << " Operations are ~" << speedup << "x faster than the old cls_bucket_head bug" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, StatisticsAccuracy) {
+ // Verify statistics are accurate (not approximate)
+
+ std::cout << "\n=== Testing Statistics Accuracy ===" << std::endl;
+
+ const std::string user_id = "accuracy_user";
+
+ // Known values
+ const uint64_t expected_bytes = 15728652; // Exact: 12 + 5MB + 10MB
+ const uint64_t expected_objects = 3;
+
+ // Update cache
+ ASSERT_EQ(0, cache->update_user_stats(user_id, expected_bytes, expected_objects));
+
+ // Retrieve and verify exact match
+ auto stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(stats.has_value());
+
+ std::cout << "Expected: " << expected_bytes << " bytes, " << expected_objects << " objects" << std::endl;
+ std::cout << "Cached: " << stats->bytes_used << " bytes, " << stats->num_objects << " objects" << std::endl;
+
+ // For in-memory cache, should be exact
+ EXPECT_EQ(expected_bytes, stats->bytes_used) << "Byte count should be exact";
+ EXPECT_EQ(expected_objects, stats->num_objects) << "Object count should be exact";
+
+ std::cout << " Statistics are exact (no approximation)" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, BackgroundRefreshSimulation) {
+ // Simulate background refresh updating expired entries
+
+ std::cout << "\n=== Simulating Background Refresh ===" << std::endl;
+
+ const std::string bucket_name = "refresh_bucket";
+
+ // Initial state
+ std::cout << "Initial upload: 1MB, 10 objects" << std::endl;
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 1024 * 1024, 10));
+
+ auto stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value());
+ EXPECT_EQ(1024 * 1024u, stats->bytes_used);
+ std::cout << " Initial stats cached" << std::endl;
+
+ // Wait for expiry
+ std::cout << "Waiting for entry to expire..." << std::endl;
+ std::this_thread::sleep_for(std::chrono::milliseconds(2500));
+
+ // Entry should be expired
+ stats = cache->get_bucket_stats(bucket_name);
+ EXPECT_FALSE(stats.has_value()) << "Entry should be expired";
+ std::cout << " Entry expired as expected" << std::endl;
+
+ // Simulate background refresh with updated stats
+ std::cout << "Simulating background refresh: 2MB, 20 objects" << std::endl;
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 2 * 1024 * 1024, 20));
+
+ // Should now have fresh data
+ stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value());
+ EXPECT_EQ(2 * 1024 * 1024u, stats->bytes_used);
+ EXPECT_EQ(20u, stats->num_objects);
+ std::cout << " Background refresh successful - stats updated" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, ConcurrentAccessSimulation) {
+ // Simulate concurrent access from multiple operations
+ // (Like multiple s3cmd operations happening simultaneously)
+
+ std::cout << "\n=== Simulating Concurrent Access ===" << std::endl;
+
+ const int num_threads = 10;
+ const int operations_per_thread = 100;
+
+ // Pre-populate some buckets
+ for (int i = 0; i < num_threads; ++i) {
+ std::string bucket_name = "concurrent_bucket_" + std::to_string(i);
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, i * 1024, i));
+ }
+
+ std::cout << "Launching " << num_threads << " threads, "
+ << operations_per_thread << " operations each..." << std::endl;
+
+ std::atomic<int> success_count{0};
+ std::atomic<int> failure_count{0};
+ std::vector<std::thread> threads;
+
+ auto start = std::chrono::high_resolution_clock::now();
+
+ for (int t = 0; t < num_threads; ++t) {
+ threads.emplace_back([this, t, operations_per_thread, &success_count, &failure_count]() {
+ std::string bucket_name = "concurrent_bucket_" + std::to_string(t);
+
+ for (int i = 0; i < operations_per_thread; ++i) {
+ auto stats = cache->get_bucket_stats(bucket_name);
+ if (stats.has_value()) {
+ success_count++;
+ } else {
+ failure_count++;
+ }
+ }
+ });
+ }
+
+ // Wait for all threads
+ for (auto& thread : threads) {
+ thread.join();
+ }
+
+ auto end = std::chrono::high_resolution_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+
+ int total_ops = num_threads * operations_per_thread;
+ std::cout << "\nConcurrent Access Results:" << std::endl;
+ std::cout << " Total operations: " << total_ops << std::endl;
+ std::cout << " Successful: " << success_count << std::endl;
+ std::cout << " Failed: " << failure_count << std::endl;
+ std::cout << " Time: " << duration.count() << " ms" << std::endl;
+ std::cout << " Average: " << (static_cast<double>(duration.count()) / total_ops)
+ << " ms/op" << std::endl;
+
+ // All should succeed
+ EXPECT_EQ(total_ops, success_count.load()) << "All operations should succeed";
+ EXPECT_EQ(0, failure_count.load()) << "No operations should fail";
+
+ std::cout << " Cache handled concurrent access successfully" << std::endl;
+}
+
+TEST_F(TestRGWUsageCache, CompleteWorkflowIntegration) {
+ // Integration test covering the complete manual test workflow
+
+ std::cout << "\n=== Complete Workflow Integration Test ===" << std::endl;
+ std::cout << "Simulating entire manual test procedure...\n" << std::endl;
+
+ // Step 1: Setup
+ std::cout << "[Step 1] Creating user and buckets" << std::endl;
+ const std::string user_id = "testuser";
+ const std::string bucket1 = "bucket1";
+ const std::string bucket2 = "bucket2";
+
+ // Step 2: Upload files
+ std::cout << "[Step 2] Uploading files" << std::endl;
+ std::cout << " bucket1: small.txt + medium.bin" << std::endl;
+ std::cout << " bucket2: large.bin" << std::endl;
+
+ uint64_t b1_bytes = 12 + (5 * 1024 * 1024);
+ uint64_t b2_bytes = 10 * 1024 * 1024;
+ uint64_t user_bytes = b1_bytes + b2_bytes;
+
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket1, b1_bytes, 2));
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket2, b2_bytes, 1));
+ ASSERT_EQ(0, cache->update_user_stats(user_id, user_bytes, 3));
+
+ // Step 3: Verify user stats (like checking perf counters)
+ std::cout << "[Step 3] Verifying user statistics" << std::endl;
+ auto user_stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(user_stats.has_value());
+ std::cout << " used_bytes: " << user_stats->bytes_used << std::endl;
+ std::cout << " num_objects: " << user_stats->num_objects << std::endl;
+ EXPECT_EQ(user_bytes, user_stats->bytes_used);
+ EXPECT_EQ(3u, user_stats->num_objects);
+
+ // Step 4: Verify bucket stats
+ std::cout << "[Step 4] Verifying bucket statistics" << std::endl;
+ auto b1_stats = cache->get_bucket_stats(bucket1);
+ ASSERT_TRUE(b1_stats.has_value());
+ std::cout << " bucket1 used_bytes: " << b1_stats->bytes_used << std::endl;
+ std::cout << " bucket1 num_objects: " << b1_stats->num_objects << std::endl;
+ EXPECT_EQ(b1_bytes, b1_stats->bytes_used);
+ EXPECT_EQ(2u, b1_stats->num_objects);
+
+ // Step 5: Test cache hits (multiple listings)
+ std::cout << "[Step 5] Testing cache hit behavior" << std::endl;
+ for (int i = 0; i < 3; ++i) {
+ auto stats = cache->get_bucket_stats(bucket1);
+ ASSERT_TRUE(stats.has_value());
+ }
+ std::cout << " Performed 3 bucket listings - all from cache" << std::endl;
+
+ // Step 6: Verify cache size
+ std::cout << "[Step 6] Verifying cache metrics" << std::endl;
+ size_t cache_size = cache->get_cache_size();
+ std::cout << " cache_size: " << cache_size << std::endl;
+ EXPECT_GE(cache_size, 3u); // user + 2 buckets
+
+ // Step 7: Performance check
+ std::cout << "[Step 7] Performance regression check" << std::endl;
+ auto start = std::chrono::high_resolution_clock::now();
+ for (int i = 0; i < 100; ++i) {
+ cache->get_bucket_stats(bucket1);
+ }
+ auto end = std::chrono::high_resolution_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+ double avg = static_cast<double>(duration.count()) / 100.0;
+ std::cout << " 100 operations: " << duration.count() << " ms (avg: " << avg << " ms/op)" << std::endl;
+ EXPECT_LT(avg, 10.0);
+
+ std::cout << "\n=== Complete Workflow Test PASSED ===" << std::endl;
+ std::cout << "All manual test scenarios validated!" << std::endl;
+}
+
// Main function
int main(int argc, char **argv) {
// Initialize Google Test
EXPECT_EQ(50.0, hit_rate); // 1 hit / 2 total = 50%
}
+// Add these test cases to the existing TestRGWUsagePerfCounters test fixture
+
+TEST_F(TestRGWUsagePerfCounters, MultipleUserBucketScenario) {
+ // This simulates a manual scenario test: multiple users with multiple buckets
+ // User: testuser with bucket1 (small.txt + medium.bin) and bucket2 (large.bin)
+
+ std::cout << "\n=== Testing Multiple User/Bucket Scenario ===" << std::endl;
+
+ const std::string user_id = "testuser";
+ const std::string bucket1 = "bucket1";
+ const std::string bucket2 = "bucket2";
+
+ // Simulate uploading to bucket1: small.txt (~12 bytes) + medium.bin (5MB)
+ const uint64_t bucket1_bytes = 12 + (5 * 1024 * 1024);
+ const uint64_t bucket1_objects = 2;
+
+ // Simulate uploading to bucket2: large.bin (10MB)
+ const uint64_t bucket2_bytes = 10 * 1024 * 1024;
+ const uint64_t bucket2_objects = 1;
+
+ // Total user stats
+ const uint64_t total_user_bytes = bucket1_bytes + bucket2_bytes;
+ const uint64_t total_user_objects = bucket1_objects + bucket2_objects;
+
+ // Update bucket stats
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket1, bucket1_bytes, bucket1_objects));
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket2, bucket2_bytes, bucket2_objects));
+
+ // Update user stats (aggregated from buckets)
+ ASSERT_EQ(0, cache->update_user_stats(user_id, total_user_bytes, total_user_objects));
+
+ std::cout << "Updated stats:" << std::endl;
+ std::cout << " User: " << user_id << std::endl;
+ std::cout << " Bucket1: " << bucket1 << " - " << bucket1_bytes << " bytes, "
+ << bucket1_objects << " objects" << std::endl;
+ std::cout << " Bucket2: " << bucket2 << " - " << bucket2_bytes << " bytes, "
+ << bucket2_objects << " objects" << std::endl;
+
+ // Verify bucket1 stats
+ auto b1_stats = cache->get_bucket_stats(bucket1);
+ ASSERT_TRUE(b1_stats.has_value()) << "Bucket1 stats not found";
+ EXPECT_EQ(bucket1_bytes, b1_stats->bytes_used);
+ EXPECT_EQ(bucket1_objects, b1_stats->num_objects);
+ std::cout << " Bucket1 stats verified" << std::endl;
+
+ // Verify bucket2 stats
+ auto b2_stats = cache->get_bucket_stats(bucket2);
+ ASSERT_TRUE(b2_stats.has_value()) << "Bucket2 stats not found";
+ EXPECT_EQ(bucket2_bytes, b2_stats->bytes_used);
+ EXPECT_EQ(bucket2_objects, b2_stats->num_objects);
+ std::cout << " Bucket2 stats verified" << std::endl;
+
+ // Verify user stats (should be sum of all buckets)
+ auto user_stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(user_stats.has_value()) << "User stats not found";
+ EXPECT_EQ(total_user_bytes, user_stats->bytes_used);
+ EXPECT_EQ(total_user_objects, user_stats->num_objects);
+ std::cout << " User stats verified: " << total_user_bytes << " bytes (~15MB), "
+ << total_user_objects << " objects" << std::endl;
+
+ // Verify expected totals match manual test expectations
+ // Expected: ~15MB total (5MB + 10MB + small file)
+ const uint64_t expected_min_bytes = 15 * 1024 * 1024; // 15MB
+ EXPECT_GE(user_stats->bytes_used, expected_min_bytes)
+ << "User should have at least 15MB";
+ EXPECT_EQ(3u, user_stats->num_objects)
+ << "User should have exactly 3 objects";
+
+ std::cout << " All statistics match expected values from manual test" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, CacheHitOnRepeatedAccess) {
+ // This simulates: s3cmd ls s3://bucket1 (multiple times)
+ // Should see cache hits increase
+
+ std::cout << "\n=== Testing Cache Hits on Repeated Access ===" << std::endl;
+
+ const std::string bucket_name = "bucket1";
+
+ // First, populate the cache
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 5 * 1024 * 1024, 2));
+
+ // Get initial hit/miss counts
+ uint64_t initial_hits = cache->get_cache_hits();
+ uint64_t initial_misses = cache->get_cache_misses();
+
+ std::cout << "Initial state - Hits: " << initial_hits
+ << ", Misses: " << initial_misses << std::endl;
+
+ // Simulate multiple bucket listings (like s3cmd ls s3://bucket1)
+ const int num_listings = 3;
+ for (int i = 0; i < num_listings; ++i) {
+ auto stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value()) << "Failed to get bucket stats on iteration " << i;
+ std::cout << " Listing " << (i + 1) << ": Found bucket with "
+ << stats->bytes_used << " bytes" << std::endl;
+ }
+
+ // Get final hit/miss counts
+ uint64_t final_hits = cache->get_cache_hits();
+ uint64_t final_misses = cache->get_cache_misses();
+
+ uint64_t hits_increase = final_hits - initial_hits;
+
+ std::cout << "Final state - Hits: " << final_hits
+ << ", Misses: " << final_misses << std::endl;
+ std::cout << "Cache hits increased by: " << hits_increase << std::endl;
+
+ // We expect all accesses to be hits since the entry is already cached
+ EXPECT_EQ(num_listings, hits_increase)
+ << "Expected " << num_listings << " cache hits, got " << hits_increase;
+
+ // Hit rate should be good
+ double hit_rate = cache->get_hit_rate();
+ std::cout << " Cache hit rate: " << hit_rate << "%" << std::endl;
+ EXPECT_GT(hit_rate, 0.0) << "Cache hit rate should be > 0%";
+
+ std::cout << " Cache behavior verified - repeated access produces cache hits" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, PerformanceNoBlockingInIOPath) {
+ // CRITICAL TEST: Verify that cache operations are fast
+ // This simulates the fix for the 90ms cls_bucket_head() issue
+
+ std::cout << "\n=== Testing Performance: No Blocking in I/O Path ===" << std::endl;
+
+ const int num_operations = 100;
+ const std::string bucket_prefix = "perf_bucket_";
+
+ // Warm up - add some entries to cache
+ for (int i = 0; i < 10; ++i) {
+ std::string bucket_name = bucket_prefix + std::to_string(i);
+ cache->update_bucket_stats(bucket_name, i * 1024, i);
+ }
+
+ // Measure time for 100 get operations
+ auto start = std::chrono::high_resolution_clock::now();
+
+ for (int i = 0; i < num_operations; ++i) {
+ std::string bucket_name = bucket_prefix + std::to_string(i % 10);
+ auto stats = cache->get_bucket_stats(bucket_name);
+ // These should be cache hits, should be VERY fast
+ }
+
+ auto end = std::chrono::high_resolution_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
+
+ double avg_per_op_us = static_cast<double>(duration.count()) / num_operations;
+ double avg_per_op_ms = avg_per_op_us / 1000.0;
+
+ std::cout << "Performance results:" << std::endl;
+ std::cout << " Total operations: " << num_operations << std::endl;
+ std::cout << " Total time: " << duration.count() << " μs ("
+ << (duration.count() / 1000.0) << " ms)" << std::endl;
+ std::cout << " Average per operation: " << avg_per_op_us << " μs ("
+ << avg_per_op_ms << " ms)" << std::endl;
+
+ // CRITICAL ASSERTION: Operations should be MUCH faster than 90ms
+ // Even with margin for test overhead, should be < 10ms per op
+ EXPECT_LT(avg_per_op_ms, 10.0)
+ << "Average operation time " << avg_per_op_ms
+ << "ms is too high - should be < 10ms (was 90ms with sync_owner_stats bug)";
+
+ // Better yet, should be sub-millisecond for cache hits
+ EXPECT_LT(avg_per_op_ms, 1.0)
+ << "Average operation time " << avg_per_op_ms
+ << "ms should be < 1ms for cache hits";
+
+ // Calculate cache hit rate - should be high since we're reusing buckets
+ double hit_rate = cache->get_hit_rate();
+ std::cout << " Cache hit rate: " << hit_rate << "%" << std::endl;
+
+ std::cout << " Performance test PASSED - no blocking in I/O path" << std::endl;
+ std::cout << " Operations are ~" << (90.0 / avg_per_op_ms) << "x faster than the old bug!" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, CacheSizeAndUpdateTracking) {
+ // This validates the cache size and update counters
+ // Simulates checking: cache_size, cache_updates counters
+
+ std::cout << "\n=== Testing Cache Size and Update Tracking ===" << std::endl;
+
+ // Initial state
+ uint64_t initial_size = cache->get_cache_size();
+ std::cout << "Initial cache size: " << initial_size << std::endl;
+ EXPECT_EQ(0u, initial_size) << "Cache should start empty";
+
+ // Add users and buckets
+ const int num_users = 3;
+ const int num_buckets = 2;
+
+ for (int i = 0; i < num_users; ++i) {
+ std::string user_id = "user_" + std::to_string(i);
+ ASSERT_EQ(0, cache->update_user_stats(user_id, (i + 1) * 1024 * 1024, (i + 1) * 10));
+ }
+
+ for (int i = 0; i < num_buckets; ++i) {
+ std::string bucket_name = "bucket_" + std::to_string(i);
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, (i + 1) * 512 * 1024, (i + 1) * 5));
+ }
+
+ // Check cache size
+ uint64_t final_size = cache->get_cache_size();
+ std::cout << "Final cache size: " << final_size << std::endl;
+
+ EXPECT_EQ(num_users + num_buckets, final_size)
+ << "Cache size should reflect " << (num_users + num_buckets) << " entries";
+
+ std::cout << " Cache size tracking verified" << std::endl;
+
+ // Verify we can retrieve all entries
+ for (int i = 0; i < num_users; ++i) {
+ std::string user_id = "user_" + std::to_string(i);
+ auto stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(stats.has_value()) << "User " << user_id << " not found in cache";
+ EXPECT_EQ((i + 1) * 1024 * 1024u, stats->bytes_used);
+ }
+
+ for (int i = 0; i < num_buckets; ++i) {
+ std::string bucket_name = "bucket_" + std::to_string(i);
+ auto stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value()) << "Bucket " << bucket_name << " not found in cache";
+ EXPECT_EQ((i + 1) * 512 * 1024u, stats->bytes_used);
+ }
+
+ std::cout << " All cache entries verified" << std::endl;
+
+ // Check that cache hits occurred
+ uint64_t hits = cache->get_cache_hits();
+ std::cout << "Total cache hits: " << hits << std::endl;
+ EXPECT_GT(hits, 0u) << "Should have some cache hits from retrievals";
+
+ std::cout << " Cache metrics tracking verified" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, StatsAccuracyWithinTolerance) {
+ // Verify that statistics are accurate within acceptable tolerance
+
+ std::cout << "\n=== Testing Statistics Accuracy ===" << std::endl;
+
+ const std::string user_id = "accuracy_test_user";
+
+ // Upload files with known sizes (matching our manual test)
+ struct FileUpload {
+ std::string name;
+ uint64_t size;
+ };
+
+ std::vector<FileUpload> files = {
+ {"small.txt", 12}, // "Hello World\n"
+ {"medium.bin", 5 * 1024 * 1024}, // 5MB
+ {"large.bin", 10 * 1024 * 1024} // 10MB
+ };
+
+ uint64_t total_bytes = 0;
+ uint64_t total_objects = files.size();
+
+ std::cout << "Simulating file uploads:" << std::endl;
+ for (const auto& file : files) {
+ total_bytes += file.size;
+ std::cout << " " << file.name << ": " << file.size << " bytes" << std::endl;
+ }
+
+ std::cout << "Total: " << total_bytes << " bytes ("
+ << (total_bytes / (1024.0 * 1024.0)) << " MB), "
+ << total_objects << " objects" << std::endl;
+
+ // Update cache with total stats
+ ASSERT_EQ(0, cache->update_user_stats(user_id, total_bytes, total_objects));
+
+ // Retrieve and verify
+ auto stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(stats.has_value()) << "User stats not found";
+
+ std::cout << "\nVerifying accuracy:" << std::endl;
+ std::cout << " Expected bytes: " << total_bytes << std::endl;
+ std::cout << " Cached bytes: " << stats->bytes_used << std::endl;
+ std::cout << " Expected objects: " << total_objects << std::endl;
+ std::cout << " Cached objects: " << stats->num_objects << std::endl;
+
+ // Exact match for in-memory cache
+ EXPECT_EQ(total_bytes, stats->bytes_used) << "Byte count should be exact";
+ EXPECT_EQ(total_objects, stats->num_objects) << "Object count should be exact";
+
+ // Verify totals match expectations (~15MB)
+ const uint64_t expected_mb = 15;
+ const uint64_t expected_bytes = expected_mb * 1024 * 1024;
+ const double tolerance = 0.1; // 10% tolerance
+
+ double bytes_diff = std::abs(static_cast<double>(stats->bytes_used) - expected_bytes);
+ double bytes_diff_pct = (bytes_diff / expected_bytes) * 100.0;
+
+ std::cout << " Difference from expected ~15MB: " << bytes_diff_pct << "%" << std::endl;
+
+ EXPECT_LT(bytes_diff_pct, tolerance * 100.0)
+ << "Byte count should be within " << (tolerance * 100.0) << "% of expected";
+
+ std::cout << " Statistics accuracy verified (within tolerance)" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, ExpiryAndRefreshScenario) {
+ // Test that expired entries are handled correctly
+ // Simulates the background refresh mechanism
+
+ std::cout << "\n=== Testing Expiry and Refresh Scenario ===" << std::endl;
+
+ const std::string bucket_name = "expiry_test_bucket";
+
+ // Add bucket stats
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 1024 * 1024, 10));
+
+ // Verify stats exist
+ auto stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value()) << "Stats should exist initially";
+ std::cout << " Initial stats cached" << std::endl;
+
+ // Access should be a cache hit
+ uint64_t hits_before = cache->get_cache_hits();
+ stats = cache->get_bucket_stats(bucket_name);
+ uint64_t hits_after = cache->get_cache_hits();
+ EXPECT_EQ(hits_before + 1, hits_after) << "Should have one more cache hit";
+ std::cout << " Cache hit recorded" << std::endl;
+
+ // Wait for TTL to expire (2 seconds + buffer)
+ std::cout << "Waiting for cache entry to expire (2.5s)..." << std::endl;
+ std::this_thread::sleep_for(std::chrono::milliseconds(2500));
+
+ // Access expired entry - should be a miss
+ uint64_t misses_before = cache->get_cache_misses();
+ stats = cache->get_bucket_stats(bucket_name);
+ uint64_t misses_after = cache->get_cache_misses();
+
+ EXPECT_FALSE(stats.has_value()) << "Expired entry should not be returned";
+ EXPECT_EQ(misses_before + 1, misses_after) << "Should have one more cache miss";
+ std::cout << " Expired entry handled correctly (cache miss)" << std::endl;
+
+ // Simulate background refresh - re-add the stats
+ std::cout << "Simulating background refresh..." << std::endl;
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket_name, 2 * 1024 * 1024, 20));
+
+ // Now should be in cache again with new values
+ stats = cache->get_bucket_stats(bucket_name);
+ ASSERT_TRUE(stats.has_value()) << "Refreshed stats should exist";
+ EXPECT_EQ(2 * 1024 * 1024u, stats->bytes_used) << "Should have updated bytes";
+ EXPECT_EQ(20u, stats->num_objects) << "Should have updated objects";
+
+ std::cout << "Background refresh simulation successful" << std::endl;
+ std::cout << "Cache properly handles expiry and refresh cycle" << std::endl;
+}
+
+TEST_F(TestRGWUsagePerfCounters, ComprehensiveManualTestWorkflow) {
+ // This test combines all the test steps into one comprehensive test
+
+ std::cout << "\n=== Comprehensive Manual Test Workflow ===" << std::endl;
+ std::cout << "Simulating complete manual testing scenario..." << std::endl;
+
+ // Step 1: Create user and buckets
+ std::cout << "\n[Step 1] Creating user and buckets..." << std::endl;
+ const std::string user_id = "testuser";
+ const std::string bucket1 = "bucket1";
+ const std::string bucket2 = "bucket2";
+
+ // Step 2: Upload files
+ std::cout << "[Step 2] Uploading files..." << std::endl;
+
+ // Bucket1: small.txt + medium.bin
+ uint64_t b1_bytes = 12 + (5 * 1024 * 1024); // ~5MB
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket1, b1_bytes, 2));
+ std::cout << " Uploaded to " << bucket1 << ": 2 files, " << b1_bytes << " bytes" << std::endl;
+
+ // Bucket2: large.bin
+ uint64_t b2_bytes = 10 * 1024 * 1024; // 10MB
+ ASSERT_EQ(0, cache->update_bucket_stats(bucket2, b2_bytes, 1));
+ std::cout << " Uploaded to " << bucket2 << ": 1 file, " << b2_bytes << " bytes" << std::endl;
+
+ // User total
+ uint64_t user_bytes = b1_bytes + b2_bytes;
+ uint64_t user_objects = 3;
+ ASSERT_EQ(0, cache->update_user_stats(user_id, user_bytes, user_objects));
+ std::cout << " User total: " << user_objects << " objects, " << user_bytes << " bytes" << std::endl;
+
+ // Step 3: Check user statistics
+ std::cout << "\n[Step 3] Checking user statistics..." << std::endl;
+ auto user_stats = cache->get_user_stats(user_id);
+ ASSERT_TRUE(user_stats.has_value());
+ std::cout << " rgw_user_" << user_id << ":" << std::endl;
+ std::cout << " used_bytes: " << user_stats->bytes_used << std::endl;
+ std::cout << " num_objects: " << user_stats->num_objects << std::endl;
+ EXPECT_EQ(user_bytes, user_stats->bytes_used);
+ EXPECT_EQ(3u, user_stats->num_objects);
+ std::cout << " User stats verified" << std::endl;
+
+ // Step 4: Check bucket statistics
+ std::cout << "\n[Step 4] Checking bucket statistics..." << std::endl;
+ auto b1_stats = cache->get_bucket_stats(bucket1);
+ ASSERT_TRUE(b1_stats.has_value());
+ std::cout << " rgw_bucket_" << bucket1 << ":" << std::endl;
+ std::cout << " used_bytes: " << b1_stats->bytes_used << std::endl;
+ std::cout << " num_objects: " << b1_stats->num_objects << std::endl;
+ EXPECT_EQ(b1_bytes, b1_stats->bytes_used);
+ EXPECT_EQ(2u, b1_stats->num_objects);
+ std::cout << " Bucket1 stats verified" << std::endl;
+
+ // Step 5: Test cache behavior (multiple listings)
+ std::cout << "\n[Step 5] Testing cache behavior..." << std::endl;
+ uint64_t hits_before = cache->get_cache_hits();
+
+ for (int i = 0; i < 3; ++i) {
+ cache->get_bucket_stats(bucket1);
+ }
+
+ uint64_t hits_after = cache->get_cache_hits();
+ uint64_t hit_increase = hits_after - hits_before;
+ std::cout << " Performed 3 bucket listings" << std::endl;
+ std::cout << " Cache hits increased by: " << hit_increase << std::endl;
+ EXPECT_GE(hit_increase, 3u);
+ std::cout << " Cache hits verified" << std::endl;
+
+ // Step 6: Check cache metrics
+ std::cout << "\n[Step 6] Checking cache metrics..." << std::endl;
+ uint64_t cache_size = cache->get_cache_size();
+ double hit_rate = cache->get_hit_rate();
+ std::cout << " Cache size: " << cache_size << std::endl;
+ std::cout << " Cache hit rate: " << hit_rate << "%" << std::endl;
+ EXPECT_GE(cache_size, 3u); // At least user + 2 buckets
+ EXPECT_GT(hit_rate, 0.0);
+ std::cout << " Cache metrics verified" << std::endl;
+
+ // Step 7: Performance check
+ std::cout << "\n[Step 7] Performance regression check..." << std::endl;
+ auto start = std::chrono::high_resolution_clock::now();
+ for (int i = 0; i < 100; ++i) {
+ cache->get_bucket_stats(bucket1);
+ }
+ auto end = std::chrono::high_resolution_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+ double avg_ms = static_cast<double>(duration.count()) / 100.0;
+
+ std::cout << " 100 operations in " << duration.count() << "ms" << std::endl;
+ std::cout << " Average: " << avg_ms << "ms per operation" << std::endl;
+ EXPECT_LT(avg_ms, 10.0) << "Should be much faster than 90ms";
+ std::cout << " No performance regression detected" << std::endl;
+
+ std::cout << "\n=== Comprehensive Test PASSED ===" << std::endl;
+ std::cout << "All manual test scenarios validated successfully!" << std::endl;
+}
+
int main(int argc, char **argv) {
::testing::InitGoogleTest(&argc, argv);
projects / ceph-ci.git / commitdiff