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
, rgw_bucket
& bucket
, RGWStorageStats
& stats
) = 0;
66 virtual bool map_find(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) = 0;
68 virtual bool map_find_and_update(const rgw_user
& user
, rgw_bucket
& bucket
, typename lru_map
<T
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) = 0;
69 virtual void map_add(const rgw_user
& user
, 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
, 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
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
, RGWStorageStats
& stats
);
86 int async_refresh(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
);
87 void async_refresh_response(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
);
89 class AsyncRefreshHandler
{
92 RGWQuotaCache
<T
> *cache
;
94 AsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<T
> *_cache
) : store(_store
), cache(_cache
) {}
95 virtual ~AsyncRefreshHandler() {}
97 virtual int init_fetch() = 0;
98 virtual void drop_reference() = 0;
101 virtual AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, rgw_bucket
& bucket
) = 0;
105 bool RGWQuotaCache
<T
>::can_use_cached_stats(RGWQuotaInfo
& quota
, RGWStorageStats
& cached_stats
)
107 if (quota
.max_size
>= 0) {
108 if (quota
.max_size_soft_threshold
< 0) {
109 quota
.max_size_soft_threshold
= quota
.max_size
* store
->ctx()->_conf
->rgw_bucket_quota_soft_threshold
;
112 const auto cached_stats_num_kb_rounded
= rgw_rounded_kb(cached_stats
.size_rounded
);
113 if (cached_stats_num_kb_rounded
>= (uint64_t)quota
.max_size_soft_threshold
) {
114 ldout(store
->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (size): "
115 << cached_stats_num_kb_rounded
<< " >= " << quota
.max_size_soft_threshold
<< dendl
;
120 if (quota
.max_objects
>= 0) {
121 if (quota
.max_objs_soft_threshold
< 0) {
122 quota
.max_objs_soft_threshold
= quota
.max_objects
* store
->ctx()->_conf
->rgw_bucket_quota_soft_threshold
;
125 if (cached_stats
.num_objects
>= (uint64_t)quota
.max_objs_soft_threshold
) {
126 ldout(store
->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (num objs): "
127 << cached_stats
.num_objects
<< " >= " << quota
.max_objs_soft_threshold
<< dendl
;
136 int RGWQuotaCache
<T
>::async_refresh(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
)
138 /* protect against multiple updates */
139 StatsAsyncTestSet test_update
;
140 if (!map_find_and_update(user
, bucket
, &test_update
)) {
141 /* most likely we just raced with another update */
145 async_refcount
->get();
148 AsyncRefreshHandler
*handler
= allocate_refresh_handler(user
, bucket
);
150 int ret
= handler
->init_fetch();
152 async_refcount
->put();
153 handler
->drop_reference();
161 void RGWQuotaCache
<T
>::async_refresh_response(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
)
163 ldout(store
->ctx(), 20) << "async stats refresh response for bucket=" << bucket
<< dendl
;
165 RGWQuotaCacheStats qs
;
167 map_find(user
, bucket
, qs
);
169 set_stats(user
, bucket
, qs
, stats
);
171 async_refcount
->put();
175 void RGWQuotaCache
<T
>::set_stats(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
, RGWStorageStats
& stats
)
178 qs
.expiration
= ceph_clock_now();
179 qs
.async_refresh_time
= qs
.expiration
;
180 qs
.expiration
+= store
->ctx()->_conf
->rgw_bucket_quota_ttl
;
181 qs
.async_refresh_time
+= store
->ctx()->_conf
->rgw_bucket_quota_ttl
/ 2;
183 map_add(user
, bucket
, qs
);
187 int RGWQuotaCache
<T
>::get_stats(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
, RGWQuotaInfo
& quota
) {
188 RGWQuotaCacheStats qs
;
189 utime_t now
= ceph_clock_now();
190 if (map_find(user
, bucket
, qs
)) {
191 if (qs
.async_refresh_time
.sec() > 0 && now
>= qs
.async_refresh_time
) {
192 int r
= async_refresh(user
, bucket
, qs
);
194 ldout(store
->ctx(), 0) << "ERROR: quota async refresh returned ret=" << r
<< dendl
;
196 /* continue processing, might be a transient error, async refresh is just optimization */
200 if (can_use_cached_stats(quota
, qs
.stats
) && qs
.expiration
>
207 int ret
= fetch_stats_from_storage(user
, bucket
, stats
);
208 if (ret
< 0 && ret
!= -ENOENT
)
211 set_stats(user
, bucket
, qs
, stats
);
218 class RGWQuotaStatsUpdate
: public lru_map
<T
, RGWQuotaCacheStats
>::UpdateContext
{
219 const int objs_delta
;
220 const uint64_t added_bytes
;
221 const uint64_t removed_bytes
;
223 RGWQuotaStatsUpdate(const int objs_delta
,
224 const uint64_t added_bytes
,
225 const uint64_t removed_bytes
)
226 : objs_delta(objs_delta
),
227 added_bytes(added_bytes
),
228 removed_bytes(removed_bytes
) {
231 bool update(RGWQuotaCacheStats
* const entry
) override
{
232 const uint64_t rounded_added
= rgw_rounded_objsize(added_bytes
);
233 const uint64_t rounded_removed
= rgw_rounded_objsize(removed_bytes
);
235 entry
->stats
.size
+= added_bytes
- removed_bytes
;
236 entry
->stats
.size_rounded
+= rounded_added
- rounded_removed
;
237 entry
->stats
.num_objects
+= objs_delta
;
245 void RGWQuotaCache
<T
>::adjust_stats(const rgw_user
& user
, rgw_bucket
& bucket
, int objs_delta
,
246 uint64_t added_bytes
, uint64_t removed_bytes
)
248 RGWQuotaStatsUpdate
<T
> update(objs_delta
, added_bytes
, removed_bytes
);
249 map_find_and_update(user
, bucket
, &update
);
251 data_modified(user
, bucket
);
254 class BucketAsyncRefreshHandler
: public RGWQuotaCache
<rgw_bucket
>::AsyncRefreshHandler
,
255 public RGWGetBucketStats_CB
{
258 BucketAsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<rgw_bucket
> *_cache
,
259 const rgw_user
& _user
, rgw_bucket
& _bucket
) :
260 RGWQuotaCache
<rgw_bucket
>::AsyncRefreshHandler(_store
, _cache
),
261 RGWGetBucketStats_CB(_bucket
), user(_user
) {}
263 void drop_reference() override
{ put(); }
264 void handle_response(int r
) override
;
265 int init_fetch() override
;
268 int BucketAsyncRefreshHandler::init_fetch()
270 RGWBucketInfo bucket_info
;
272 RGWObjectCtx
obj_ctx(store
);
274 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
276 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
280 ldout(store
->ctx(), 20) << "initiating async quota refresh for bucket=" << bucket
<< dendl
;
282 r
= store
->get_bucket_stats_async(bucket_info
, RGW_NO_SHARD
, this);
284 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
.name
<< dendl
;
286 /* get_bucket_stats_async() dropped our reference already */
293 void BucketAsyncRefreshHandler::handle_response(const int r
)
296 ldout(store
->ctx(), 20) << "AsyncRefreshHandler::handle_response() r=" << r
<< dendl
;
297 return; /* nothing to do here */
302 for (const auto& pair
: *stats
) {
303 const RGWStorageStats
& s
= pair
.second
;
306 bs
.size_rounded
+= s
.size_rounded
;
307 bs
.num_objects
+= s
.num_objects
;
310 cache
->async_refresh_response(user
, bucket
, bs
);
313 class RGWBucketStatsCache
: public RGWQuotaCache
<rgw_bucket
> {
315 bool map_find(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
316 return stats_map
.find(bucket
, qs
);
319 bool map_find_and_update(const rgw_user
& user
, rgw_bucket
& bucket
, lru_map
<rgw_bucket
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) override
{
320 return stats_map
.find_and_update(bucket
, NULL
, ctx
);
323 void map_add(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
324 stats_map
.add(bucket
, qs
);
327 int fetch_stats_from_storage(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
) override
;
330 explicit RGWBucketStatsCache(RGWRados
*_store
) : RGWQuotaCache
<rgw_bucket
>(_store
, _store
->ctx()->_conf
->rgw_bucket_quota_cache_size
) {
333 AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, rgw_bucket
& bucket
) override
{
334 return new BucketAsyncRefreshHandler(store
, this, user
, bucket
);
338 int RGWBucketStatsCache::fetch_stats_from_storage(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
)
340 RGWBucketInfo bucket_info
;
342 RGWObjectCtx
obj_ctx(store
);
344 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
346 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
353 map
<RGWObjCategory
, RGWStorageStats
> bucket_stats
;
354 r
= store
->get_bucket_stats(bucket_info
, RGW_NO_SHARD
, &bucket_ver
,
355 &master_ver
, bucket_stats
, nullptr);
357 ldout(store
->ctx(), 0) << "could not get bucket stats for bucket="
358 << bucket
.name
<< dendl
;
362 stats
= RGWStorageStats();
364 for (const auto& pair
: bucket_stats
) {
365 const RGWStorageStats
& s
= pair
.second
;
367 stats
.size
+= s
.size
;
368 stats
.size_rounded
+= s
.size_rounded
;
369 stats
.num_objects
+= s
.num_objects
;
375 class UserAsyncRefreshHandler
: public RGWQuotaCache
<rgw_user
>::AsyncRefreshHandler
,
376 public RGWGetUserStats_CB
{
379 UserAsyncRefreshHandler(RGWRados
*_store
, RGWQuotaCache
<rgw_user
> *_cache
,
380 const rgw_user
& _user
, rgw_bucket
& _bucket
) :
381 RGWQuotaCache
<rgw_user
>::AsyncRefreshHandler(_store
, _cache
),
382 RGWGetUserStats_CB(_user
),
385 void drop_reference() override
{ put(); }
386 int init_fetch() override
;
387 void handle_response(int r
) override
;
390 int UserAsyncRefreshHandler::init_fetch()
392 ldout(store
->ctx(), 20) << "initiating async quota refresh for user=" << user
<< dendl
;
393 int r
= store
->get_user_stats_async(user
, this);
395 ldout(store
->ctx(), 0) << "could not get bucket info for user=" << user
<< dendl
;
397 /* get_bucket_stats_async() dropped our reference already */
404 void UserAsyncRefreshHandler::handle_response(int r
)
407 ldout(store
->ctx(), 20) << "AsyncRefreshHandler::handle_response() r=" << r
<< dendl
;
408 return; /* nothing to do here */
411 cache
->async_refresh_response(user
, bucket
, stats
);
414 class RGWUserStatsCache
: public RGWQuotaCache
<rgw_user
> {
415 std::atomic
<bool> down_flag
= { false };
417 map
<rgw_bucket
, rgw_user
> modified_buckets
;
419 /* thread, sync recent modified buckets info */
420 class BucketsSyncThread
: public Thread
{
422 RGWUserStatsCache
*stats
;
428 BucketsSyncThread(CephContext
*_cct
, RGWUserStatsCache
*_s
) : cct(_cct
), stats(_s
), lock("RGWUserStatsCache::BucketsSyncThread") {}
430 void *entry() override
{
431 ldout(cct
, 20) << "BucketsSyncThread: start" << dendl
;
433 map
<rgw_bucket
, rgw_user
> buckets
;
435 stats
->swap_modified_buckets(buckets
);
437 for (map
<rgw_bucket
, rgw_user
>::iterator iter
= buckets
.begin(); iter
!= buckets
.end(); ++iter
) {
438 rgw_bucket bucket
= iter
->first
;
439 rgw_user
& user
= iter
->second
;
440 ldout(cct
, 20) << "BucketsSyncThread: sync user=" << user
<< " bucket=" << bucket
<< dendl
;
441 int r
= stats
->sync_bucket(user
, bucket
);
443 ldout(cct
, 0) << "WARNING: sync_bucket() returned r=" << r
<< dendl
;
447 if (stats
->going_down())
451 cond
.WaitInterval(lock
, utime_t(cct
->_conf
->rgw_user_quota_bucket_sync_interval
, 0));
453 } while (!stats
->going_down());
454 ldout(cct
, 20) << "BucketsSyncThread: done" << dendl
;
460 Mutex::Locker
l(lock
);
466 * thread, full sync all users stats periodically
468 * only sync non idle users or ones that never got synced before, this is needed so that
469 * users that didn't have quota turned on before (or existed before the user objclass
470 * tracked stats) need to get their backend stats up to date.
472 class UserSyncThread
: public Thread
{
474 RGWUserStatsCache
*stats
;
480 UserSyncThread(CephContext
*_cct
, RGWUserStatsCache
*_s
) : cct(_cct
), stats(_s
), lock("RGWUserStatsCache::UserSyncThread") {}
482 void *entry() override
{
483 ldout(cct
, 20) << "UserSyncThread: start" << dendl
;
485 int ret
= stats
->sync_all_users();
487 ldout(cct
, 5) << "ERROR: sync_all_users() returned ret=" << ret
<< dendl
;
491 cond
.WaitInterval(lock
, utime_t(cct
->_conf
->rgw_user_quota_sync_interval
, 0));
493 } while (!stats
->going_down());
494 ldout(cct
, 20) << "UserSyncThread: done" << dendl
;
500 Mutex::Locker
l(lock
);
505 BucketsSyncThread
*buckets_sync_thread
;
506 UserSyncThread
*user_sync_thread
;
508 bool map_find(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
509 return stats_map
.find(user
, qs
);
512 bool map_find_and_update(const rgw_user
& user
, rgw_bucket
& bucket
, lru_map
<rgw_user
, RGWQuotaCacheStats
>::UpdateContext
*ctx
) override
{
513 return stats_map
.find_and_update(user
, NULL
, ctx
);
516 void map_add(const rgw_user
& user
, rgw_bucket
& bucket
, RGWQuotaCacheStats
& qs
) override
{
517 stats_map
.add(user
, qs
);
520 int fetch_stats_from_storage(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
) override
;
521 int sync_bucket(const rgw_user
& rgw_user
, rgw_bucket
& bucket
);
522 int sync_user(const rgw_user
& user
);
523 int sync_all_users();
525 void data_modified(const rgw_user
& user
, rgw_bucket
& bucket
) override
;
527 void swap_modified_buckets(map
<rgw_bucket
, rgw_user
>& out
) {
529 modified_buckets
.swap(out
);
533 template<class T
> /* easier doing it as a template, Thread doesn't have ->stop() */
534 void stop_thread(T
**pthr
) {
546 RGWUserStatsCache(RGWRados
*_store
, bool quota_threads
) : RGWQuotaCache
<rgw_user
>(_store
, _store
->ctx()->_conf
->rgw_bucket_quota_cache_size
),
547 rwlock("RGWUserStatsCache::rwlock") {
549 buckets_sync_thread
= new BucketsSyncThread(store
->ctx(), this);
550 buckets_sync_thread
->create("rgw_buck_st_syn");
551 user_sync_thread
= new UserSyncThread(store
->ctx(), this);
552 user_sync_thread
->create("rgw_user_st_syn");
554 buckets_sync_thread
= NULL
;
555 user_sync_thread
= NULL
;
558 ~RGWUserStatsCache() override
{
562 AsyncRefreshHandler
*allocate_refresh_handler(const rgw_user
& user
, rgw_bucket
& bucket
) override
{
563 return new UserAsyncRefreshHandler(store
, this, user
, bucket
);
566 bool can_use_cached_stats(RGWQuotaInfo
& quota
, RGWStorageStats
& stats
) override
{
567 /* in the user case, the cached stats may contain a better estimation of the totals, as
568 * the backend is only periodically getting updated.
580 stop_thread(&buckets_sync_thread
);
582 stop_thread(&user_sync_thread
);
586 int RGWUserStatsCache::fetch_stats_from_storage(const rgw_user
& user
, rgw_bucket
& bucket
, RGWStorageStats
& stats
)
588 int r
= store
->get_user_stats(user
, stats
);
590 ldout(store
->ctx(), 0) << "could not get user stats for user=" << user
<< dendl
;
597 int RGWUserStatsCache::sync_bucket(const rgw_user
& user
, rgw_bucket
& bucket
)
599 RGWBucketInfo bucket_info
;
601 RGWObjectCtx
obj_ctx(store
);
603 int r
= store
->get_bucket_instance_info(obj_ctx
, bucket
, bucket_info
, NULL
, NULL
);
605 ldout(store
->ctx(), 0) << "could not get bucket info for bucket=" << bucket
<< " r=" << r
<< dendl
;
609 r
= rgw_bucket_sync_user_stats(store
, user
, bucket_info
);
611 ldout(store
->ctx(), 0) << "ERROR: rgw_bucket_sync_user_stats() for user=" << user
<< ", bucket=" << bucket
<< " returned " << r
<< dendl
;
618 int RGWUserStatsCache::sync_user(const rgw_user
& user
)
620 cls_user_header header
;
621 string user_str
= user
.to_str();
622 int ret
= store
->cls_user_get_header(user_str
, &header
);
624 ldout(store
->ctx(), 5) << "ERROR: can't read user header: ret=" << ret
<< dendl
;
628 if (!store
->ctx()->_conf
->rgw_user_quota_sync_idle_users
&&
629 header
.last_stats_update
< header
.last_stats_sync
) {
630 ldout(store
->ctx(), 20) << "user is idle, not doing a full sync (user=" << user
<< ")" << dendl
;
634 real_time when_need_full_sync
= header
.last_stats_sync
;
635 when_need_full_sync
+= make_timespan(store
->ctx()->_conf
->rgw_user_quota_sync_wait_time
);
637 // check if enough time passed since last full sync
638 /* FIXME: missing check? */
640 ret
= rgw_user_sync_all_stats(store
, user
);
642 ldout(store
->ctx(), 0) << "ERROR: failed user stats sync, ret=" << ret
<< dendl
;
649 int RGWUserStatsCache::sync_all_users()
654 int ret
= store
->meta_mgr
->list_keys_init(key
, &handle
);
656 ldout(store
->ctx(), 10) << "ERROR: can't get key: ret=" << ret
<< dendl
;
665 ret
= store
->meta_mgr
->list_keys_next(handle
, max
, keys
, &truncated
);
667 ldout(store
->ctx(), 0) << "ERROR: lists_keys_next(): ret=" << ret
<< dendl
;
670 for (list
<string
>::iterator iter
= keys
.begin();
671 iter
!= keys
.end() && !going_down();
673 rgw_user
user(*iter
);
674 ldout(store
->ctx(), 20) << "RGWUserStatsCache: sync user=" << user
<< dendl
;
675 int ret
= sync_user(user
);
677 ldout(store
->ctx(), 5) << "ERROR: sync_user() failed, user=" << user
<< " ret=" << ret
<< dendl
;
679 /* continuing to next user */
687 store
->meta_mgr
->list_keys_complete(handle
);
691 void RGWUserStatsCache::data_modified(const rgw_user
& user
, rgw_bucket
& bucket
)
693 /* racy, but it's ok */
695 bool need_update
= modified_buckets
.find(bucket
) == modified_buckets
.end();
700 modified_buckets
[bucket
] = user
;
706 class RGWQuotaInfoApplier
{
707 /* NOTE: no non-static field allowed as instances are supposed to live in
708 * the static memory only. */
710 RGWQuotaInfoApplier() = default;
713 virtual ~RGWQuotaInfoApplier() {}
715 virtual bool is_size_exceeded(const char * const entity
,
716 const RGWQuotaInfo
& qinfo
,
717 const RGWStorageStats
& stats
,
718 const uint64_t size
) const = 0;
720 virtual bool is_num_objs_exceeded(const char * const entity
,
721 const RGWQuotaInfo
& qinfo
,
722 const RGWStorageStats
& stats
,
723 const uint64_t num_objs
) const = 0;
725 static const RGWQuotaInfoApplier
& get_instance(const RGWQuotaInfo
& qinfo
);
728 class RGWQuotaInfoDefApplier
: public RGWQuotaInfoApplier
{
730 bool is_size_exceeded(const char * const entity
,
731 const RGWQuotaInfo
& qinfo
,
732 const RGWStorageStats
& stats
,
733 const uint64_t size
) const override
;
735 bool is_num_objs_exceeded(const char * const entity
,
736 const RGWQuotaInfo
& qinfo
,
737 const RGWStorageStats
& stats
,
738 const uint64_t num_objs
) const override
;
741 class RGWQuotaInfoRawApplier
: public RGWQuotaInfoApplier
{
743 bool is_size_exceeded(const char * const entity
,
744 const RGWQuotaInfo
& qinfo
,
745 const RGWStorageStats
& stats
,
746 const uint64_t size
) const override
;
748 bool is_num_objs_exceeded(const char * const entity
,
749 const RGWQuotaInfo
& qinfo
,
750 const RGWStorageStats
& stats
,
751 const uint64_t num_objs
) const override
;
755 bool RGWQuotaInfoDefApplier::is_size_exceeded(const char * const entity
,
756 const RGWQuotaInfo
& qinfo
,
757 const RGWStorageStats
& stats
,
758 const uint64_t size
) const
760 if (qinfo
.max_size
< 0) {
761 /* The limit is not enabled. */
765 const uint64_t cur_size
= stats
.size_rounded
;
766 const uint64_t new_size
= rgw_rounded_objsize(size
);
768 if (cur_size
+ new_size
> static_cast<uint64_t>(qinfo
.max_size
)) {
769 dout(10) << "quota exceeded: stats.size_rounded=" << stats
.size_rounded
770 << " size=" << new_size
<< " "
771 << entity
<< "_quota.max_size=" << qinfo
.max_size
<< dendl
;
778 bool RGWQuotaInfoDefApplier::is_num_objs_exceeded(const char * const entity
,
779 const RGWQuotaInfo
& qinfo
,
780 const RGWStorageStats
& stats
,
781 const uint64_t num_objs
) const
783 if (qinfo
.max_objects
< 0) {
784 /* The limit is not enabled. */
788 if (stats
.num_objects
+ num_objs
> static_cast<uint64_t>(qinfo
.max_objects
)) {
789 dout(10) << "quota exceeded: stats.num_objects=" << stats
.num_objects
790 << " " << entity
<< "_quota.max_objects=" << qinfo
.max_objects
798 bool RGWQuotaInfoRawApplier::is_size_exceeded(const char * const entity
,
799 const RGWQuotaInfo
& qinfo
,
800 const RGWStorageStats
& stats
,
801 const uint64_t size
) const
803 if (qinfo
.max_size
< 0) {
804 /* The limit is not enabled. */
808 const uint64_t cur_size
= stats
.size
;
810 if (cur_size
+ size
> static_cast<uint64_t>(qinfo
.max_size
)) {
811 dout(10) << "quota exceeded: stats.size=" << stats
.size
812 << " size=" << size
<< " "
813 << entity
<< "_quota.max_size=" << qinfo
.max_size
<< dendl
;
820 bool RGWQuotaInfoRawApplier::is_num_objs_exceeded(const char * const entity
,
821 const RGWQuotaInfo
& qinfo
,
822 const RGWStorageStats
& stats
,
823 const uint64_t num_objs
) const
825 if (qinfo
.max_objects
< 0) {
826 /* The limit is not enabled. */
830 if (stats
.num_objects
+ num_objs
> static_cast<uint64_t>(qinfo
.max_objects
)) {
831 dout(10) << "quota exceeded: stats.num_objects=" << stats
.num_objects
832 << " " << entity
<< "_quota.max_objects=" << qinfo
.max_objects
840 const RGWQuotaInfoApplier
& RGWQuotaInfoApplier::get_instance(
841 const RGWQuotaInfo
& qinfo
)
843 static RGWQuotaInfoDefApplier default_qapplier
;
844 static RGWQuotaInfoRawApplier raw_qapplier
;
846 if (qinfo
.check_on_raw
) {
849 return default_qapplier
;
854 class RGWQuotaHandlerImpl
: public RGWQuotaHandler
{
856 RGWBucketStatsCache bucket_stats_cache
;
857 RGWUserStatsCache user_stats_cache
;
859 int check_quota(const char * const entity
,
860 const RGWQuotaInfo
& quota
,
861 const RGWStorageStats
& stats
,
862 const uint64_t num_objs
,
863 const uint64_t size
) {
864 if (!quota
.enabled
) {
868 const auto& quota_applier
= RGWQuotaInfoApplier::get_instance(quota
);
870 ldout(store
->ctx(), 20) << entity
871 << " quota: max_objects=" << quota
.max_objects
872 << " max_size=" << quota
.max_size
<< dendl
;
875 if (quota_applier
.is_num_objs_exceeded(entity
, quota
, stats
, num_objs
)) {
876 return -ERR_QUOTA_EXCEEDED
;
879 if (quota_applier
.is_size_exceeded(entity
, quota
, stats
, size
)) {
880 return -ERR_QUOTA_EXCEEDED
;
883 ldout(store
->ctx(), 20) << entity
<< " quota OK:"
884 << " stats.num_objects=" << stats
.num_objects
885 << " stats.size=" << stats
.size
<< dendl
;
889 RGWQuotaHandlerImpl(RGWRados
*_store
, bool quota_threads
) : store(_store
),
890 bucket_stats_cache(_store
),
891 user_stats_cache(_store
, quota_threads
) {}
893 int check_quota(const rgw_user
& user
,
895 RGWQuotaInfo
& user_quota
,
896 RGWQuotaInfo
& bucket_quota
,
898 uint64_t size
) override
{
900 if (!bucket_quota
.enabled
&& !user_quota
.enabled
) {
905 * we need to fetch bucket stats if the user quota is enabled, because
906 * the whole system relies on us periodically updating the user's bucket
907 * stats in the user's header, this happens in get_stats() if we actually
908 * fetch that info and not rely on cached data
911 if (bucket_quota
.enabled
) {
912 RGWStorageStats bucket_stats
;
913 int ret
= bucket_stats_cache
.get_stats(user
, bucket
, bucket_stats
,
918 ret
= check_quota("bucket", bucket_quota
, bucket_stats
, num_objs
, size
);
924 if (user_quota
.enabled
) {
925 RGWStorageStats user_stats
;
926 int ret
= user_stats_cache
.get_stats(user
, bucket
, user_stats
, user_quota
);
930 ret
= check_quota("user", user_quota
, user_stats
, num_objs
, size
);
938 void update_stats(const rgw_user
& user
, rgw_bucket
& bucket
, int obj_delta
, uint64_t added_bytes
, uint64_t removed_bytes
) override
{
939 bucket_stats_cache
.adjust_stats(user
, bucket
, obj_delta
, added_bytes
, removed_bytes
);
940 user_stats_cache
.adjust_stats(user
, bucket
, obj_delta
, added_bytes
, removed_bytes
);
945 RGWQuotaHandler
*RGWQuotaHandler::generate_handler(RGWRados
*store
, bool quota_threads
)
947 return new RGWQuotaHandlerImpl(store
, quota_threads
);
950 void RGWQuotaHandler::free_handler(RGWQuotaHandler
*handler
)