rgw/logging: add quota enforcement to bucket logging

Signed-off-by: Yuval Lifshitz <ylifshit@ibm.com>

diff --git a/src/rgw/rgw_bucket_logging.cc b/src/rgw/rgw_bucket_logging.cc
index a9710dee996cf6443906cd2c5ccfc4b6edb8a68a..d26128e1f89ba4181d2433ac1acd1e80eeed0e2d 100644 (file)
--- a/src/rgw/rgw_bucket_logging.cc
+++ b/src/rgw/rgw_bucket_logging.cc
@@ -305,7 +305,12 @@ int commit_logging_object(const configuration& conf,
       target_bucket->get_key() << "'. ret = " << ret << dendl;
     return ret;
   }
-  return target_bucket->commit_logging_object(obj_name, y, dpp);
+  if (const auto ret = target_bucket->commit_logging_object(obj_name, y, dpp); ret <0 ) {
+    ldpp_dout(dpp, 1) << "ERROR: failed to commit logging object '" << obj_name << "' of bucket '" <<
+      target_bucket->get_key() << "'. ret = " << ret << dendl;
+    return ret;
+  }
+  return 0;
 }
 
 int rollover_logging_object(const configuration& conf,
@@ -548,6 +553,32 @@ int log_record(rgw::sal::Driver* driver,
       return -EINVAL;
   }
 
+  // get quota of the owner of the target bucket
+  RGWQuota user_quota;
+  if (ret = get_owner_quota_info(dpp, y, driver, target_bucket->get_owner(), user_quota); ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: failed to get quota of owner of target logging bucket '" <<
+      target_bucket_id << "' failed. ret = " << ret << dendl;
+    return ret;
+  }
+  // start with system default quota
+  // and combine with the user quota
+  RGWQuota quota;
+  driver->get_quota(quota);
+  if (target_bucket->get_info().quota.enabled) {
+    quota.bucket_quota = target_bucket->get_info().quota;
+  } else if (user_quota.bucket_quota.enabled) {
+    quota.bucket_quota = user_quota.bucket_quota;
+  }
+  if (user_quota.user_quota.enabled) {
+    quota.user_quota = user_quota.user_quota;
+  }
+  // verify there is enough quota to write the record
+  if (ret = target_bucket->check_quota(dpp, quota, record.length(), y); ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: quota check on target logging bucket '" <<
+      target_bucket_id << "' failed. ret = " << ret << dendl;
+    return ret;
+  }
+
   if (ret = target_bucket->write_logging_object(obj_name,
         record,
         y,

diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 750345c45bf8082de65cb06c3aa7ea0ab933b8bc..88804c118e3cd607f54aa2c4cc3e23d0650fbcd6 100644 (file)
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -1496,7 +1496,7 @@ int RGWOp::do_aws4_auth_completion()
   return 0;
 }
 
-static int get_owner_quota_info(DoutPrefixProvider* dpp,
+int get_owner_quota_info(const DoutPrefixProvider* dpp,
                                 optional_yield y,
                                 rgw::sal::Driver* driver,
                                 const rgw_owner& owner,

diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index cbe441c140bfdf5883812ad6964108a7e367026f..eb1468712140f414ec0a8f0a12a22c2d73a05e21 100644 (file)
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -88,6 +88,12 @@ std::tuple<bool, bool> rgw_check_policy_condition(const DoutPrefixProvider *dpp,
 
 int rgw_iam_add_buckettags(const DoutPrefixProvider *dpp, req_state* s);
 
+int get_owner_quota_info(const DoutPrefixProvider* dpp,
+                                optional_yield y,
+                                rgw::sal::Driver* driver,
+                                const rgw_owner& owner,
+                                RGWQuota& quotas);
+
 class RGWHandler {
 protected:
   rgw::sal::Driver* driver{nullptr};