1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2013 Inktank, Inc
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
16 #include "include/utime.h"
17 #include "common/lru_map.h"
18 #include "common/RefCountedObj.h"
19 #include "common/Thread.h"
20 #include "common/Mutex.h"
21 #include "common/RWLock.h"
23 #include "rgw_common.h"
24 #include "rgw_rados.h"
25 #include "rgw_quota.h"
26 #include "rgw_bucket.h"
31 #define dout_context g_ceph_context
32 #define dout_subsys ceph_subsys_rgw
35 struct RGWQuotaCacheStats
{
36 RGWStorageStats stats
;
38 utime_t async_refresh_time
;
45 lru_map
<T
, RGWQuotaCacheStats
> stats_map
;
46 RefCountedWaitObject
*async_refcount
;
48 class StatsAsyncTestSet
: public lru_map
<T
, RGWQuotaCacheStats
>::UpdateContext
{
51 uint64_t removed_bytes
;
53 StatsAsyncTestSet() : objs_delta(0), added_bytes(0), removed_bytes(0) {}
54 bool update(RGWQuotaCacheStats
*entry
) override
{
55 if (entry
->async_refresh_time
.sec() == 0)
58 entry
->async_refresh_time
= utime_t(0, 0);
64 virtual int fetch_stats_from_storage(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
) = 0;
66 virtual bool map_find(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) = 0;
68 virtual bool map_find_and_update(const rgw_user
& user
, const rgw_bucket
& bucket
, typename lru_map
<T
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) = 0;
69 virtual void map_add(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) = 0;
71 virtual void data_modified(const rgw_user
& user
, rgw_bucket
& bucket
) {}
73 RGWQuotaCache(RGWRados
*_store
, int size
) : store(_store
), stats_map(size
) {
74 async_refcount
= new RefCountedWaitObject
;
76 virtual ~RGWQuotaCache() {
77 async_refcount
->put_wait(); /* wait for all pending async requests to complete */
80 int get_stats(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
, RGWQuotaInfo
& quota
);
81 void adjust_stats(const rgw_user
& user
, rgw_bucket
& bucket
, int objs_delta
, uint64_t added_bytes
, uint64_t removed_bytes
);
83 virtual bool can_use_cached_stats(RGWQuotaInfo
& quota
, RGWStorageStats
& stats
);
85 void set_stats(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
, RGWStorageStats
& stats
);
86 int async_refresh(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
);
87 void async_refresh_response(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
);
88 void async_refresh_fail(const rgw_user
& user
, rgw_bucket
& bucket
);
90 class AsyncRefreshHandler
{
93 RGWQuotaCache
<T
> *cache
;
95 AsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<T
> *_cache
) : store(_store
), cache(_cache
) {}
96 virtual ~AsyncRefreshHandler() {}
98 virtual int init_fetch() = 0;
99 virtual void drop_reference() = 0;
102 virtual AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, const rgw_bucket
& bucket
) = 0;
106 bool RGWQuotaCache
<T
>::can_use_cached_stats(RGWQuotaInfo
& quota
, RGWStorageStats
& cached_stats
)
108 if (quota
.max_size
>= 0) {
109 if (quota
.max_size_soft_threshold
< 0) {
110 quota
.max_size_soft_threshold
= quota
.max_size
* store
->ctx()->_conf
->rgw_bucket_quota_soft_threshold
;
113 const auto cached_stats_num_kb_rounded
= rgw_rounded_kb(cached_stats
.size_rounded
);
114 if (cached_stats_num_kb_rounded
>= (uint64_t)quota
.max_size_soft_threshold
) {
115 ldout(store
->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (size): "
116 << cached_stats_num_kb_rounded
<< " >= " << quota
.max_size_soft_threshold
<< dendl
;
121 if (quota
.max_objects
>= 0) {
122 if (quota
.max_objs_soft_threshold
< 0) {
123 quota
.max_objs_soft_threshold
= quota
.max_objects
* store
->ctx()->_conf
->rgw_bucket_quota_soft_threshold
;
126 if (cached_stats
.num_objects
>= (uint64_t)quota
.max_objs_soft_threshold
) {
127 ldout(store
->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (num objs): "
128 << cached_stats
.num_objects
<< " >= " << quota
.max_objs_soft_threshold
<< dendl
;
137 int RGWQuotaCache
<T
>::async_refresh(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
)
139 /* protect against multiple updates */
140 StatsAsyncTestSet test_update
;
141 if (!map_find_and_update(user
, bucket
, &test_update
)) {
142 /* most likely we just raced with another update */
146 async_refcount
->get();
149 AsyncRefreshHandler
*handler
= allocate_refresh_handler(user
, bucket
);
151 int ret
= handler
->init_fetch();
153 async_refcount
->put();
154 handler
->drop_reference();
162 void RGWQuotaCache
<T
>::async_refresh_fail(const rgw_user
& user
, rgw_bucket
& bucket
)
164 ldout(store
->ctx(), 20) << "async stats refresh response for bucket=" << bucket
<< dendl
;
166 async_refcount
->put();
170 void RGWQuotaCache
<T
>::async_refresh_response(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
)
172 ldout(store
->ctx(), 20) << "async stats refresh response for bucket=" << bucket
<< dendl
;
174 RGWQuotaCacheStats qs
;
176 map_find(user
, bucket
, qs
);
178 set_stats(user
, bucket
, qs
, stats
);
180 async_refcount
->put();
184 void RGWQuotaCache
<T
>::set_stats(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
, RGWStorageStats
& stats
)
187 qs
.expiration
= ceph_clock_now();
188 qs
.async_refresh_time
= qs
.expiration
;
189 qs
.expiration
+= store
->ctx()->_conf
->rgw_bucket_quota_ttl
;
190 qs
.async_refresh_time
+= store
->ctx()->_conf
->rgw_bucket_quota_ttl
/ 2;
192 map_add(user
, bucket
, qs
);
196 int RGWQuotaCache
<T
>::get_stats(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
, RGWQuotaInfo
& quota
) {
197 RGWQuotaCacheStats qs
;
198 utime_t now
= ceph_clock_now();
199 if (map_find(user
, bucket
, qs
)) {
200 if (qs
.async_refresh_time
.sec() > 0 && now
>= qs
.async_refresh_time
) {
201 int r
= async_refresh(user
, bucket
, qs
);
203 ldout(store
->ctx(), 0) << "ERROR: quota async refresh returned ret=" << r
<< dendl
;
205 /* continue processing, might be a transient error, async refresh is just optimization */
209 if (can_use_cached_stats(quota
, qs
.stats
) && qs
.expiration
>
216 int ret
= fetch_stats_from_storage(user
, bucket
, stats
);
217 if (ret
< 0 && ret
!= -ENOENT
)
220 set_stats(user
, bucket
, qs
, stats
);
227 class RGWQuotaStatsUpdate
: public lru_map
<T
, RGWQuotaCacheStats
>::UpdateContext
{
228 const int objs_delta
;
229 const uint64_t added_bytes
;
230 const uint64_t removed_bytes
;
232 RGWQuotaStatsUpdate(const int objs_delta
,
233 const uint64_t added_bytes
,
234 const uint64_t removed_bytes
)
235 : objs_delta(objs_delta
),
236 added_bytes(added_bytes
),
237 removed_bytes(removed_bytes
) {
240 bool update(RGWQuotaCacheStats
* const entry
) override
{
241 const uint64_t rounded_added
= rgw_rounded_objsize(added_bytes
);
242 const uint64_t rounded_removed
= rgw_rounded_objsize(removed_bytes
);
244 if ((entry
->stats
.size
+ added_bytes
- removed_bytes
) >= 0) {
245 entry
->stats
.size
+= added_bytes
- removed_bytes
;
247 entry
->stats
.size
= 0;
250 if ((entry
->stats
.size_rounded
+ rounded_added
- rounded_removed
) >= 0) {
251 entry
->stats
.size_rounded
+= rounded_added
- rounded_removed
;
253 entry
->stats
.size_rounded
= 0;
256 if ((entry
->stats
.num_objects
+ objs_delta
) >= 0) {
257 entry
->stats
.num_objects
+= objs_delta
;
259 entry
->stats
.num_objects
= 0;
268 void RGWQuotaCache
<T
>::adjust_stats(const rgw_user
& user
, rgw_bucket
& bucket
, int objs_delta
,
269 uint64_t added_bytes
, uint64_t removed_bytes
)
271 RGWQuotaStatsUpdate
<T
> update(objs_delta
, added_bytes
, removed_bytes
);
272 map_find_and_update(user
, bucket
, &update
);
274 data_modified(user
, bucket
);
277 class BucketAsyncRefreshHandler
: public RGWQuotaCache
<rgw_bucket
>::AsyncRefreshHandler
,
278 public RGWGetBucketStats_CB
{
281 BucketAsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<rgw_bucket
> *_cache
,
282 const rgw_user
& _user
, const rgw_bucket
& _bucket
) :
283 RGWQuotaCache
<rgw_bucket
>::AsyncRefreshHandler(_store
, _cache
),
284 RGWGetBucketStats_CB(_bucket
), user(_user
) {}
286 void drop_reference() override
{ put(); }
287 void handle_response(int r
) override
;
288 int init_fetch() override
;
291 int BucketAsyncRefreshHandler::init_fetch()
293 RGWBucketInfo bucket_info
;
295 RGWObjectCtx
obj_ctx(store
);
297 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
299 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
303 ldout(store
->ctx(), 20) << "initiating async quota refresh for bucket=" << bucket
<< dendl
;
305 r
= store
->get_bucket_stats_async(bucket_info
, RGW_NO_SHARD
, this);
307 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
.name
<< dendl
;
309 /* get_bucket_stats_async() dropped our reference already */
316 void BucketAsyncRefreshHandler::handle_response(const int r
)
319 ldout(store
->ctx(), 20) << "AsyncRefreshHandler::handle_response() r=" << r
<< dendl
;
320 cache
->async_refresh_fail(user
, bucket
);
326 for (const auto& pair
: *stats
) {
327 const RGWStorageStats
& s
= pair
.second
;
330 bs
.size_rounded
+= s
.size_rounded
;
331 bs
.num_objects
+= s
.num_objects
;
334 cache
->async_refresh_response(user
, bucket
, bs
);
337 class RGWBucketStatsCache
: public RGWQuotaCache
<rgw_bucket
> {
339 bool map_find(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
340 return stats_map
.find(bucket
, qs
);
343 bool map_find_and_update(const rgw_user
& user
, const rgw_bucket
& bucket
, lru_map
<rgw_bucket
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) override
{
344 return stats_map
.find_and_update(bucket
, NULL
, ctx
);
347 void map_add(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
348 stats_map
.add(bucket
, qs
);
351 int fetch_stats_from_storage(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
) override
;
354 explicit RGWBucketStatsCache(RGWRados
*_store
) : RGWQuotaCache
<rgw_bucket
>(_store
, _store
->ctx()->_conf
->rgw_bucket_quota_cache_size
) {
357 AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, const rgw_bucket
& bucket
) override
{
358 return new BucketAsyncRefreshHandler(store
, this, user
, bucket
);
362 int RGWBucketStatsCache::fetch_stats_from_storage(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
)
364 RGWBucketInfo bucket_info
;
366 RGWObjectCtx
obj_ctx(store
);
368 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
370 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
377 map
<RGWObjCategory
, RGWStorageStats
> bucket_stats
;
378 r
= store
->get_bucket_stats(bucket_info
, RGW_NO_SHARD
, &bucket_ver
,
379 &master_ver
, bucket_stats
, nullptr);
381 ldout(store
->ctx(), 0) << "could not get bucket stats for bucket="
382 << bucket
.name
<< dendl
;
386 stats
= RGWStorageStats();
388 for (const auto& pair
: bucket_stats
) {
389 const RGWStorageStats
& s
= pair
.second
;
391 stats
.size
+= s
.size
;
392 stats
.size_rounded
+= s
.size_rounded
;
393 stats
.num_objects
+= s
.num_objects
;
399 class UserAsyncRefreshHandler
: public RGWQuotaCache
<rgw_user
>::AsyncRefreshHandler
,
400 public RGWGetUserStats_CB
{
403 UserAsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<rgw_user
> *_cache
,
404 const rgw_user
& _user
, const rgw_bucket
& _bucket
) :
405 RGWQuotaCache
<rgw_user
>::AsyncRefreshHandler(_store
, _cache
),
406 RGWGetUserStats_CB(_user
),
409 void drop_reference() override
{ put(); }
410 int init_fetch() override
;
411 void handle_response(int r
) override
;
414 int UserAsyncRefreshHandler::init_fetch()
416 ldout(store
->ctx(), 20) << "initiating async quota refresh for user=" << user
<< dendl
;
417 int r
= store
->get_user_stats_async(user
, this);
419 ldout(store
->ctx(), 0) << "could not get bucket info for user=" << user
<< dendl
;
421 /* get_bucket_stats_async() dropped our reference already */
428 void UserAsyncRefreshHandler::handle_response(int r
)
431 ldout(store
->ctx(), 20) << "AsyncRefreshHandler::handle_response() r=" << r
<< dendl
;
432 cache
->async_refresh_fail(user
, bucket
);
436 cache
->async_refresh_response(user
, bucket
, stats
);
439 class RGWUserStatsCache
: public RGWQuotaCache
<rgw_user
> {
440 std::atomic
<bool> down_flag
= { false };
442 map
<rgw_bucket
, rgw_user
> modified_buckets
;
444 /* thread, sync recent modified buckets info */
445 class BucketsSyncThread
: public Thread
{
447 RGWUserStatsCache
*stats
;
453 BucketsSyncThread(CephContext
*_cct
, RGWUserStatsCache
*_s
) : cct(_cct
), stats(_s
), lock("RGWUserStatsCache::BucketsSyncThread") {}
455 void *entry() override
{
456 ldout(cct
, 20) << "BucketsSyncThread: start" << dendl
;
458 map
<rgw_bucket
, rgw_user
> buckets
;
460 stats
->swap_modified_buckets(buckets
);
462 for (map
<rgw_bucket
, rgw_user
>::iterator iter
= buckets
.begin(); iter
!= buckets
.end(); ++iter
) {
463 rgw_bucket bucket
= iter
->first
;
464 rgw_user
& user
= iter
->second
;
465 ldout(cct
, 20) << "BucketsSyncThread: sync user=" << user
<< " bucket=" << bucket
<< dendl
;
466 int r
= stats
->sync_bucket(user
, bucket
);
468 ldout(cct
, 0) << "WARNING: sync_bucket() returned r=" << r
<< dendl
;
472 if (stats
->going_down())
476 cond
.WaitInterval(lock
, utime_t(cct
->_conf
->rgw_user_quota_bucket_sync_interval
, 0));
478 } while (!stats
->going_down());
479 ldout(cct
, 20) << "BucketsSyncThread: done" << dendl
;
485 Mutex::Locker
l(lock
);
491 * thread, full sync all users stats periodically
493 * only sync non idle users or ones that never got synced before, this is needed so that
494 * users that didn't have quota turned on before (or existed before the user objclass
495 * tracked stats) need to get their backend stats up to date.
497 class UserSyncThread
: public Thread
{
499 RGWUserStatsCache
*stats
;
505 UserSyncThread(CephContext
*_cct
, RGWUserStatsCache
*_s
) : cct(_cct
), stats(_s
), lock("RGWUserStatsCache::UserSyncThread") {}
507 void *entry() override
{
508 ldout(cct
, 20) << "UserSyncThread: start" << dendl
;
510 int ret
= stats
->sync_all_users();
512 ldout(cct
, 5) << "ERROR: sync_all_users() returned ret=" << ret
<< dendl
;
516 cond
.WaitInterval(lock
, utime_t(cct
->_conf
->rgw_user_quota_sync_interval
, 0));
518 } while (!stats
->going_down());
519 ldout(cct
, 20) << "UserSyncThread: done" << dendl
;
525 Mutex::Locker
l(lock
);
530 BucketsSyncThread
*buckets_sync_thread
;
531 UserSyncThread
*user_sync_thread
;
533 bool map_find(const rgw_user
& user
,const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
534 return stats_map
.find(user
, qs
);
537 bool map_find_and_update(const rgw_user
& user
, const rgw_bucket
& bucket
, lru_map
<rgw_user
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) override
{
538 return stats_map
.find_and_update(user
, NULL
, ctx
);
541 void map_add(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
542 stats_map
.add(user
, qs
);
545 int fetch_stats_from_storage(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
) override
;
546 int sync_bucket(const rgw_user
& rgw_user
, rgw_bucket
& bucket
);
547 int sync_user(const rgw_user
& user
);
548 int sync_all_users();
550 void data_modified(const rgw_user
& user
, rgw_bucket
& bucket
) override
;
552 void swap_modified_buckets(map
<rgw_bucket
, rgw_user
>& out
) {
554 modified_buckets
.swap(out
);
558 template<class T
> /* easier doing it as a template, Thread doesn't have ->stop() */
559 void stop_thread(T
**pthr
) {
571 RGWUserStatsCache(RGWRados
*_store
, bool quota_threads
) : RGWQuotaCache
<rgw_user
>(_store
, _store
->ctx()->_conf
->rgw_bucket_quota_cache_size
),
572 rwlock("RGWUserStatsCache::rwlock") {
574 buckets_sync_thread
= new BucketsSyncThread(store
->ctx(), this);
575 buckets_sync_thread
->create("rgw_buck_st_syn");
576 user_sync_thread
= new UserSyncThread(store
->ctx(), this);
577 user_sync_thread
->create("rgw_user_st_syn");
579 buckets_sync_thread
= NULL
;
580 user_sync_thread
= NULL
;
583 ~RGWUserStatsCache() override
{
587 AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, const rgw_bucket
& bucket
) override
{
588 return new UserAsyncRefreshHandler(store
, this, user
, bucket
);
591 bool can_use_cached_stats(RGWQuotaInfo
& quota
, RGWStorageStats
& stats
) override
{
592 /* in the user case, the cached stats may contain a better estimation of the totals, as
593 * the backend is only periodically getting updated.
605 stop_thread(&buckets_sync_thread
);
607 stop_thread(&user_sync_thread
);
611 int RGWUserStatsCache::fetch_stats_from_storage(const rgw_user
& user
, const rgw_bucket
& bucket
, RGWStorageStats
& stats
)
613 int r
= store
->get_user_stats(user
, stats
);
615 ldout(store
->ctx(), 0) << "could not get user stats for user=" << user
<< dendl
;
622 int RGWUserStatsCache::sync_bucket(const rgw_user
& user
, rgw_bucket
& bucket
)
624 RGWBucketInfo bucket_info
;
626 RGWObjectCtx
obj_ctx(store
);
628 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
630 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
634 r
= rgw_bucket_sync_user_stats(store
, user
, bucket_info
);
636 ldout(store
->ctx(), 0) << "ERROR: rgw_bucket_sync_user_stats() for user=" << user
<< ", bucket=" << bucket
<< " returned " << r
<< dendl
;
643 int RGWUserStatsCache::sync_user(const rgw_user
& user
)
645 cls_user_header header
;
646 string user_str
= user
.to_str();
647 int ret
= store
->cls_user_get_header(user_str
, &header
);
649 ldout(store
->ctx(), 5) << "ERROR: can't read user header: ret=" << ret
<< dendl
;
653 if (!store
->ctx()->_conf
->rgw_user_quota_sync_idle_users
&&
654 header
.last_stats_update
< header
.last_stats_sync
) {
655 ldout(store
->ctx(), 20) << "user is idle, not doing a full sync (user=" << user
<< ")" << dendl
;
659 real_time when_need_full_sync
= header
.last_stats_sync
;
660 when_need_full_sync
+= make_timespan(store
->ctx()->_conf
->rgw_user_quota_sync_wait_time
);
662 // check if enough time passed since last full sync
663 /* FIXME: missing check? */
665 ret
= rgw_user_sync_all_stats(store
, user
);
667 ldout(store
->ctx(), 0) << "ERROR: failed user stats sync, ret=" << ret
<< dendl
;
674 int RGWUserStatsCache::sync_all_users()
679 int ret
= store
->meta_mgr
->list_keys_init(key
, &handle
);
681 ldout(store
->ctx(), 10) << "ERROR: can't get key: ret=" << ret
<< dendl
;
690 ret
= store
->meta_mgr
->list_keys_next(handle
, max
, keys
, &truncated
);
692 ldout(store
->ctx(), 0) << "ERROR: lists_keys_next(): ret=" << ret
<< dendl
;
695 for (list
<string
>::iterator iter
= keys
.begin();
696 iter
!= keys
.end() && !going_down();
698 rgw_user
user(*iter
);
699 ldout(store
->ctx(), 20) << "RGWUserStatsCache: sync user=" << user
<< dendl
;
700 int ret
= sync_user(user
);
702 ldout(store
->ctx(), 5) << "ERROR: sync_user() failed, user=" << user
<< " ret=" << ret
<< dendl
;
704 /* continuing to next user */
712 store
->meta_mgr
->list_keys_complete(handle
);
716 void RGWUserStatsCache::data_modified(const rgw_user
& user
, rgw_bucket
& bucket
)
718 /* racy, but it's ok */
720 bool need_update
= modified_buckets
.find(bucket
) == modified_buckets
.end();
725 modified_buckets
[bucket
] = user
;
731 class RGWQuotaInfoApplier
{
732 /* NOTE: no non-static field allowed as instances are supposed to live in
733 * the static memory only. */
735 RGWQuotaInfoApplier() = default;
738 virtual ~RGWQuotaInfoApplier() {}
740 virtual bool is_size_exceeded(const char * const entity
,
741 const RGWQuotaInfo
& qinfo
,
742 const RGWStorageStats
& stats
,
743 const uint64_t size
) const = 0;
745 virtual bool is_num_objs_exceeded(const char * const entity
,
746 const RGWQuotaInfo
& qinfo
,
747 const RGWStorageStats
& stats
,
748 const uint64_t num_objs
) const = 0;
750 static const RGWQuotaInfoApplier
& get_instance(const RGWQuotaInfo
& qinfo
);
753 class RGWQuotaInfoDefApplier
: public RGWQuotaInfoApplier
{
755 bool is_size_exceeded(const char * const entity
,
756 const RGWQuotaInfo
& qinfo
,
757 const RGWStorageStats
& stats
,
758 const uint64_t size
) const override
;
760 bool is_num_objs_exceeded(const char * const entity
,
761 const RGWQuotaInfo
& qinfo
,
762 const RGWStorageStats
& stats
,
763 const uint64_t num_objs
) const override
;
766 class RGWQuotaInfoRawApplier
: public RGWQuotaInfoApplier
{
768 bool is_size_exceeded(const char * const entity
,
769 const RGWQuotaInfo
& qinfo
,
770 const RGWStorageStats
& stats
,
771 const uint64_t size
) const override
;
773 bool is_num_objs_exceeded(const char * const entity
,
774 const RGWQuotaInfo
& qinfo
,
775 const RGWStorageStats
& stats
,
776 const uint64_t num_objs
) const override
;
780 bool RGWQuotaInfoDefApplier::is_size_exceeded(const char * const entity
,
781 const RGWQuotaInfo
& qinfo
,
782 const RGWStorageStats
& stats
,
783 const uint64_t size
) const
785 if (qinfo
.max_size
< 0) {
786 /* The limit is not enabled. */
790 const uint64_t cur_size
= stats
.size_rounded
;
791 const uint64_t new_size
= rgw_rounded_objsize(size
);
793 if (cur_size
+ new_size
> static_cast<uint64_t>(qinfo
.max_size
)) {
794 dout(10) << "quota exceeded: stats.size_rounded=" << stats
.size_rounded
795 << " size=" << new_size
<< " "
796 << entity
<< "_quota.max_size=" << qinfo
.max_size
<< dendl
;
803 bool RGWQuotaInfoDefApplier::is_num_objs_exceeded(const char * const entity
,
804 const RGWQuotaInfo
& qinfo
,
805 const RGWStorageStats
& stats
,
806 const uint64_t num_objs
) const
808 if (qinfo
.max_objects
< 0) {
809 /* The limit is not enabled. */
813 if (stats
.num_objects
+ num_objs
> static_cast<uint64_t>(qinfo
.max_objects
)) {
814 dout(10) << "quota exceeded: stats.num_objects=" << stats
.num_objects
815 << " " << entity
<< "_quota.max_objects=" << qinfo
.max_objects
823 bool RGWQuotaInfoRawApplier::is_size_exceeded(const char * const entity
,
824 const RGWQuotaInfo
& qinfo
,
825 const RGWStorageStats
& stats
,
826 const uint64_t size
) const
828 if (qinfo
.max_size
< 0) {
829 /* The limit is not enabled. */
833 const uint64_t cur_size
= stats
.size
;
835 if (cur_size
+ size
> static_cast<uint64_t>(qinfo
.max_size
)) {
836 dout(10) << "quota exceeded: stats.size=" << stats
.size
837 << " size=" << size
<< " "
838 << entity
<< "_quota.max_size=" << qinfo
.max_size
<< dendl
;
845 bool RGWQuotaInfoRawApplier::is_num_objs_exceeded(const char * const entity
,
846 const RGWQuotaInfo
& qinfo
,
847 const RGWStorageStats
& stats
,
848 const uint64_t num_objs
) const
850 if (qinfo
.max_objects
< 0) {
851 /* The limit is not enabled. */
855 if (stats
.num_objects
+ num_objs
> static_cast<uint64_t>(qinfo
.max_objects
)) {
856 dout(10) << "quota exceeded: stats.num_objects=" << stats
.num_objects
857 << " " << entity
<< "_quota.max_objects=" << qinfo
.max_objects
865 const RGWQuotaInfoApplier
& RGWQuotaInfoApplier::get_instance(
866 const RGWQuotaInfo
& qinfo
)
868 static RGWQuotaInfoDefApplier default_qapplier
;
869 static RGWQuotaInfoRawApplier raw_qapplier
;
871 if (qinfo
.check_on_raw
) {
874 return default_qapplier
;
879 class RGWQuotaHandlerImpl
: public RGWQuotaHandler
{
881 RGWBucketStatsCache bucket_stats_cache
;
882 RGWUserStatsCache user_stats_cache
;
884 int check_quota(const char * const entity
,
885 const RGWQuotaInfo
& quota
,
886 const RGWStorageStats
& stats
,
887 const uint64_t num_objs
,
888 const uint64_t size
) {
889 if (!quota
.enabled
) {
893 const auto& quota_applier
= RGWQuotaInfoApplier::get_instance(quota
);
895 ldout(store
->ctx(), 20) << entity
896 << " quota: max_objects=" << quota
.max_objects
897 << " max_size=" << quota
.max_size
<< dendl
;
900 if (quota_applier
.is_num_objs_exceeded(entity
, quota
, stats
, num_objs
)) {
901 return -ERR_QUOTA_EXCEEDED
;
904 if (quota_applier
.is_size_exceeded(entity
, quota
, stats
, size
)) {
905 return -ERR_QUOTA_EXCEEDED
;
908 ldout(store
->ctx(), 20) << entity
<< " quota OK:"
909 << " stats.num_objects=" << stats
.num_objects
910 << " stats.size=" << stats
.size
<< dendl
;
914 RGWQuotaHandlerImpl(RGWRados
*_store
, bool quota_threads
) : store(_store
),
915 bucket_stats_cache(_store
),
916 user_stats_cache(_store
, quota_threads
) {}
918 int check_quota(const rgw_user
& user
,
920 RGWQuotaInfo
& user_quota
,
921 RGWQuotaInfo
& bucket_quota
,
923 uint64_t size
) override
{
925 if (!bucket_quota
.enabled
&& !user_quota
.enabled
) {
930 * we need to fetch bucket stats if the user quota is enabled, because
931 * the whole system relies on us periodically updating the user's bucket
932 * stats in the user's header, this happens in get_stats() if we actually
933 * fetch that info and not rely on cached data
936 if (bucket_quota
.enabled
) {
937 RGWStorageStats bucket_stats
;
938 int ret
= bucket_stats_cache
.get_stats(user
, bucket
, bucket_stats
,
943 ret
= check_quota("bucket", bucket_quota
, bucket_stats
, num_objs
, size
);
949 if (user_quota
.enabled
) {
950 RGWStorageStats user_stats
;
951 int ret
= user_stats_cache
.get_stats(user
, bucket
, user_stats
, user_quota
);
955 ret
= check_quota("user", user_quota
, user_stats
, num_objs
, size
);
963 void update_stats(const rgw_user
& user
, rgw_bucket
& bucket
, int obj_delta
, uint64_t added_bytes
, uint64_t removed_bytes
) override
{
964 bucket_stats_cache
.adjust_stats(user
, bucket
, obj_delta
, added_bytes
, removed_bytes
);
965 user_stats_cache
.adjust_stats(user
, bucket
, obj_delta
, added_bytes
, removed_bytes
);
968 int check_bucket_shards(uint64_t max_objs_per_shard
, uint64_t num_shards
,
969 const rgw_user
& user
, const rgw_bucket
& bucket
, RGWQuotaInfo
& bucket_quota
,
970 uint64_t num_objs
, bool& need_resharding
, uint32_t *suggested_num_shards
)
972 RGWStorageStats bucket_stats
;
973 int ret
= bucket_stats_cache
.get_stats(user
, bucket
, bucket_stats
,
979 if (bucket_stats
.num_objects
+ num_objs
> num_shards
* max_objs_per_shard
) {
980 ldout(store
->ctx(), 0) << __func__
<< ": resharding needed: stats.num_objects=" << bucket_stats
.num_objects
981 << " shard max_objects=" << max_objs_per_shard
* num_shards
<< dendl
;
982 need_resharding
= true;
983 if (suggested_num_shards
) {
984 *suggested_num_shards
= (bucket_stats
.num_objects
+ num_objs
) * 2 / max_objs_per_shard
;
987 need_resharding
= false;
996 RGWQuotaHandler
*RGWQuotaHandler::generate_handler(RGWRados
*store
, bool quota_threads
)
998 return new RGWQuotaHandlerImpl(store
, quota_threads
);
1001 void RGWQuotaHandler::free_handler(RGWQuotaHandler
*handler
)