rgw/posix: implement the quota feature

Implement the quota feature for the POSIX driver.

Signed-off-by: Nithya Balachandran <nithya.balachandran@ibm.com>

diff --git a/src/rgw/driver/dbstore/common/dbstore.cc b/src/rgw/driver/dbstore/common/dbstore.cc
index bf8edcdbb2eb891dcc2d108b22a954259aeafb98..8a35ebca054c6d8cf76f7c9f7c8ed21310df3acd 100644 (file)
--- a/src/rgw/driver/dbstore/common/dbstore.cc
+++ b/src/rgw/driver/dbstore/common/dbstore.cc
@@ -104,6 +104,8 @@ std::shared_ptr<class DBOp> DB::getDBOp(const DoutPrefixProvider *dpp, std::stri
     return dbops.RemoveUser;
   if (!Op.compare("GetUser"))
     return dbops.GetUser;
+  if (!Op.compare("ListUsers"))
+    return dbops.ListUsers;
   if (!Op.compare("InsertBucket"))
     return dbops.InsertBucket;
   if (!Op.compare("UpdateBucket"))
@@ -431,6 +433,33 @@ out:
   return ret;
 }
 
+int DB::list_users(const DoutPrefixProvider *dpp,
+        const std::string& marker,
+        uint64_t max,
+        std::list<std::string>& keys,
+        bool *is_truncated)
+{
+  int ret = 0;
+  DBOpParams params = {};
+  InitializeParams(dpp, &params);
+
+  params.op.user.uinfo.user_id = marker;
+  params.op.list_max_count = max;
+
+  ret = ProcessOp(dpp, "ListUsers", &params);
+
+  if (ret) {
+    ldpp_dout(dpp, 0) << "list_users failed with err:(" << ret <<") " << dendl;
+    goto out;
+  }
+  for (auto& entry : params.op.user.list_entries) {
+    keys.push_back(entry.user_id.to_str());
+  }
+
+out:
+  return ret;
+}
+
 int DB::get_account(const DoutPrefixProvider *dpp,
     const std::string& query_str, const std::string& query_str_val,
     RGWAccountInfo& ainfo, map<string, bufferlist> *pattrs,

diff --git a/src/rgw/driver/dbstore/common/dbstore.h b/src/rgw/driver/dbstore/common/dbstore.h
index d038600b8125d4718833916870ccdc8b283da5e8..d141770d46a994cf8c8a390db5d7620bc2922614 100644 (file)
--- a/src/rgw/driver/dbstore/common/dbstore.h
+++ b/src/rgw/driver/dbstore/common/dbstore.h
@@ -36,6 +36,7 @@ struct DBOpUserInfo {
   RGWUserInfo uinfo = {};
   obj_version user_version;
   rgw::sal::Attrs user_attrs;
+  std::list<RGWUserInfo> list_entries;
 };
 
 struct DBOpBucketInfo {
@@ -372,6 +373,7 @@ struct DBOps {
   std::shared_ptr<class InsertUserOp> InsertUser;
   std::shared_ptr<class RemoveUserOp> RemoveUser;
   std::shared_ptr<class GetUserOp> GetUser;
+  std::shared_ptr<class ListUsersOp> ListUsers;
   std::shared_ptr<class InsertBucketOp> InsertBucket;
   std::shared_ptr<class UpdateBucketOp> UpdateBucket;
   std::shared_ptr<class RemoveBucketOp> RemoveBucket;
@@ -947,6 +949,28 @@ class GetUserOp: virtual public DBOp {
     }
 };
 
+
+class ListUsersOp: virtual public DBOp {
+    static constexpr std::string_view Query = "SELECT \
+                          UserID, Tenant, NS, DisplayName, UserEmail, \
+                          AccessKeysID, AccessKeysSecret, AccessKeys, SwiftKeys,\
+                          SubUsers, Suspended, MaxBuckets, OpMask, UserCaps, Admin, \
+                          System, PlacementName, PlacementStorageClass, PlacementTags, \
+                          BucketQuota, TempURLKeys, UserQuota, Type, MfaIDs, AssumedRoleARN, \
+                          UserAttrs, UserVersion, UserVersionTag from '{}' where \
+                          UserID >= {} ORDER BY UserID ASC LIMIT {} ";
+
+  public:
+    virtual ~ListUsersOp() {}
+
+    static std::string Schema(DBOpPrepareParams &params) {
+      return fmt::format(Query,
+        params.user_table,
+        params.op.user.user_id,
+        params.op.list_max_count);
+    }
+};
+
 class InsertBucketOp: virtual public DBOp {
   private:
     static constexpr std::string_view Query =
@@ -1717,6 +1741,11 @@ class DB {
         RGWObjVersionTracker *pobjv_tracker, RGWUserInfo* pold_info);
     int remove_user(const DoutPrefixProvider *dpp,
         RGWUserInfo& uinfo, RGWObjVersionTracker *pobjv_tracker);
+    int list_users(const DoutPrefixProvider *dpp,
+        const std::string& marker,
+        uint64_t max,
+        std::list<std::string>& keys,
+        bool *is_truncated);
     int get_account(const DoutPrefixProvider *dpp,
         const std::string& query_str, const std::string& query_str_val,
         RGWAccountInfo& ainfo, std::map<std::string, bufferlist> *pattrs,

diff --git a/src/rgw/driver/dbstore/sqlite/sqliteDB.cc b/src/rgw/driver/dbstore/sqlite/sqliteDB.cc
index ac573c4fd06d5a2563ec1fec3f58cae2dd3e4791..c9791b87b484e14ff3b4557fa7074eee68c598e1 100644 (file)
--- a/src/rgw/driver/dbstore/sqlite/sqliteDB.cc
+++ b/src/rgw/driver/dbstore/sqlite/sqliteDB.cc
@@ -436,6 +436,7 @@ static int list_user(const DoutPrefixProvider *dpp, DBOpInfo &op, sqlite3_stmt *
   op.user.user_version.ver = sqlite3_column_int(stmt, UserVersion);
   op.user.user_version.tag = (const char*)sqlite3_column_text(stmt, UserVersionTag);
 
+  op.user.list_entries.push_back(op.user.uinfo);
   return 0;
 }
 
@@ -630,6 +631,7 @@ int SQLiteDB::InitializeDBOps(const DoutPrefixProvider *dpp)
   dbops.InsertUser = make_shared<SQLInsertUser>(&this->db, this->getDBname(), cct);
   dbops.RemoveUser = make_shared<SQLRemoveUser>(&this->db, this->getDBname(), cct);
   dbops.GetUser = make_shared<SQLGetUser>(&this->db, this->getDBname(), cct);
+  dbops.ListUsers = make_shared<SQLListUsers>(&this->db, this->getDBname(), cct);
   dbops.InsertBucket = make_shared<SQLInsertBucket>(&this->db, this->getDBname(), cct);
   dbops.UpdateBucket = make_shared<SQLUpdateBucket>(&this->db, this->getDBname(), cct);
   dbops.RemoveBucket = make_shared<SQLRemoveBucket>(&this->db, this->getDBname(), cct);
@@ -1555,6 +1557,48 @@ out:
   return ret;
 }
 
+int SQLListUsers::Prepare(const DoutPrefixProvider *dpp, struct DBOpParams *params)
+{
+  int ret = -1;
+  struct DBOpPrepareParams p_params = PrepareParams;
+
+  if (!*sdb) {
+    ldpp_dout(dpp, 0)<<"In SQLListUsers - no db" << dendl;
+    goto out;
+  }
+
+  InitPrepareParams(dpp, p_params, params);
+
+  SQL_PREPARE(dpp, p_params, sdb, stmt, ret, "PrepareListUsers");
+out:
+  return ret;
+}
+
+int SQLListUsers::Bind(const DoutPrefixProvider *dpp, struct DBOpParams *params)
+{
+  int index = -1;
+  int rc = 0;
+  struct DBOpPrepareParams p_params = PrepareParams;
+
+  SQL_BIND_INDEX(dpp, stmt, index, p_params.op.user.user_id, sdb);
+  SQL_BIND_TEXT(dpp, stmt, index, params->op.user.uinfo.user_id.id.c_str(), sdb);
+
+  SQL_BIND_INDEX(dpp, stmt, index, p_params.op.list_max_count, sdb);
+  SQL_BIND_INT(dpp, stmt, index, params->op.list_max_count, sdb);
+out:
+  return rc;
+}
+
+int SQLListUsers::Execute(const DoutPrefixProvider *dpp, struct DBOpParams *params)
+{
+  int ret = -1;
+
+  SQL_EXECUTE(dpp, params, stmt, list_user);
+
+out:
+  return ret;
+}
+
 int SQLInsertBucket::Prepare(const DoutPrefixProvider *dpp, struct DBOpParams *params)
 {
   int ret = -1;

diff --git a/src/rgw/driver/dbstore/sqlite/sqliteDB.h b/src/rgw/driver/dbstore/sqlite/sqliteDB.h
index 342224d161411645d73a34e1c0bc47f7c9d19126..e368db70853caa770b2638c9ad7fd1ed812586fa 100644 (file)
--- a/src/rgw/driver/dbstore/sqlite/sqliteDB.h
+++ b/src/rgw/driver/dbstore/sqlite/sqliteDB.h
@@ -196,6 +196,23 @@ class SQLGetUser : public SQLiteDB, public GetUserOp {
     int Bind(const DoutPrefixProvider *dpp, DBOpParams *params);
 };
 
+class SQLListUsers : public SQLiteDB, public ListUsersOp {
+  private:
+    sqlite3 **sdb = NULL;
+    sqlite3_stmt *stmt = NULL; // Prepared statement
+
+  public:
+    SQLListUsers(void **db, std::string db_name, CephContext *cct) : SQLiteDB((sqlite3 *)(*db), db_name, cct), sdb((sqlite3 **)db) {}
+    ~SQLListUsers() {
+      if (stmt)
+        sqlite3_finalize(stmt);
+    }
+    int Prepare(const DoutPrefixProvider *dpp, DBOpParams *params);
+    int Execute(const DoutPrefixProvider *dpp, DBOpParams *params);
+    int Bind(const DoutPrefixProvider *dpp, DBOpParams *params);
+};
+
+
 class SQLInsertBucket : public SQLiteDB, public InsertBucketOp {
   private:
     sqlite3 **sdb = NULL;

diff --git a/src/rgw/driver/posix/rgw_sal_posix.cc b/src/rgw/driver/posix/rgw_sal_posix.cc
index c1cb6a7a0e80c2eacf1d7e849073782d4d7aa1c0..92710d01669166f3184a149b9eb724abb6902890 100644 (file)
--- a/src/rgw/driver/posix/rgw_sal_posix.cc
+++ b/src/rgw/driver/posix/rgw_sal_posix.cc
@@ -1971,11 +1971,17 @@ int POSIXDriver::initialize(CephContext *cct, const DoutPrefixProvider *dpp)
     }
   }
   ldpp_dout(dpp, 20) << "root_fd: " << root_dir->get_fd() << dendl;
+  quota_handler = RGWQuotaHandler::generate_handler(dpp, this, true);
 
   ldpp_dout(dpp, 20) << "SUCCESS" << dendl;
   return 0;
 }
 
+void POSIXDriver::finalize()
+{
+  RGWQuotaHandler::free_handler(quota_handler);
+}
+
 std::unique_ptr<User> POSIXDriver::get_user(const rgw_user &u)
 {
   return std::make_unique<POSIXUser>(this, u);
@@ -2305,15 +2311,22 @@ int POSIXDriver::list_buckets(const DoutPrefixProvider* dpp, const rgw_owner& ow
       errno = 0;
       continue;
     }
-
+    std::unique_ptr<Bucket> bucket;
+    ret = load_bucket(dpp, rgw_bucket("", entry->d_name), &bucket, null_yield);
+    if (bucket->get_owner() != owner) {
+      continue;
+    }
     RGWBucketEnt ent;
     ent.bucket.name = url_decode(entry->d_name);
     ent.creation_time = ceph::real_clock::from_time_t(stx.stx_btime.tv_sec);
     // TODO: ent.size and ent.count
 
     result.buckets.push_back(std::move(ent));
-
     errno = 0;
+    if (result.buckets.size() == max){
+      result.next_marker = ent.bucket.marker;
+      break;
+    }
   }
   ret = errno;
   if (ret != 0) {
@@ -2479,6 +2492,116 @@ std::unique_ptr<RGWRole> POSIXDriver::get_role(const RGWRoleInfo& info)
   return std::unique_ptr<RGWRole>(p);
 }
 
+struct meta_list_handle {
+  std::string marker;
+  std::string section;
+
+  DIR *dir = nullptr;
+  long dpos = -1;
+
+  meta_list_handle(const std::string& _section, const std::string& _marker) {
+    marker = _marker;
+    section = _section;
+  }
+};
+
+int POSIXDriver::meta_list_keys_init(const DoutPrefixProvider *dpp,
+                                     const std::string& section,
+                                     const std::string& marker, void** phandle)
+{
+  meta_list_handle* stuff = new meta_list_handle(section, marker);
+  *phandle = (void *)stuff;
+  if (section == "bucket") {
+    int ret;
+    int dfd = copy_dir_fd(get_root_fd());
+    if (dfd == -1) {
+      ret = errno;
+      ldpp_dout(dpp, 0) << "ERROR: could not open root to list buckets: "
+                        << cpp_strerror(errno) << dendl;
+      return -ret;
+    }
+
+    stuff->dir = fdopendir(dfd);
+    if (stuff->dir == NULL) {
+      ret = errno;
+      ldpp_dout(dpp, 0) << "ERROR: could not open root to list buckets: "
+                        << cpp_strerror(ret) << dendl;
+      ::close(dfd);
+      return -ret;
+    }
+  }
+  return 0;
+  }
+
+int POSIXDriver::meta_list_keys_next(const DoutPrefixProvider *dpp, void* handle,
+                                     int max, std::list<std::string>& keys,
+                                     bool* truncated)
+{
+  meta_list_handle *h = static_cast<meta_list_handle *>(handle);
+  *truncated = false;
+  int ret;
+  keys.clear();
+  if (h->section == "user") {
+    ret = get_user_db()->list_users(dpp, h->marker, max, keys, truncated);
+    if (ret < 0) {
+      return ret;
+    }
+    if (keys.size() > 0) {
+      h->marker = *keys.rbegin();
+      if (std::cmp_equal(keys.size(),max)) {
+        *truncated = true;
+      }
+    }
+  } else if (h->section == "bucket") {
+    if (h->dpos != -1) {
+      seekdir(h->dir, h->dpos);
+    }
+    struct dirent* entry;
+    while ((entry = readdir(h->dir)) != NULL) {
+      if (entry->d_type == DT_UNKNOWN) {
+        struct statx stx;
+
+        ret = statx(get_root_fd(), entry->d_name, AT_SYMLINK_NOFOLLOW, STATX_ALL, &stx);
+        if (ret < 0) {
+          ret = errno;
+          ldpp_dout(dpp, 0) << "ERROR: could not stat object " << entry->d_name << ": "
+                           << cpp_strerror(ret) << dendl;
+          return -ret;
+        }
+        if (!S_ISDIR(stx.stx_mode)) {
+        /* Not a bucket, skip it */
+          continue;
+        }
+      } else if (entry->d_type != DT_DIR) {
+        continue;
+      }
+      if (entry->d_name[0] == '.') {
+        /* Skip dotfiles */
+        continue;
+     }
+      keys.push_back(entry->d_name);
+      if (std::cmp_equal(keys.size(),max)) {
+        h->dpos = telldir(h->dir);
+        *truncated = true;
+        break;
+      }
+    }
+  }
+  return 0;
+}
+
+void POSIXDriver::meta_list_keys_complete(void* handle)
+{
+  if (handle) {
+    meta_list_handle *h = static_cast<meta_list_handle *>(handle);
+    if (h->section == "bucket") {
+      closedir(h->dir);
+    }
+    delete h;
+  }
+  return;
+}
+
 int POSIXBucket::fill_cache(const DoutPrefixProvider* dpp, optional_yield y,
                             fill_cache_cb_t& cb)
 {
@@ -2839,7 +2962,9 @@ int POSIXBucket::check_empty(const DoutPrefixProvider* dpp, optional_yield y)
 int POSIXBucket::check_quota(const DoutPrefixProvider *dpp, RGWQuota& quota, uint64_t obj_size,
                                optional_yield y, bool check_size_only)
 {
-    return 0;
+  return driver->get_quota_handler()->check_quota(dpp, info.owner, get_key(),
+                                                  quota, (check_size_only ? 0 : 1),
+                                                  obj_size, y);
 }
 
 int POSIXBucket::try_refresh_info(const DoutPrefixProvider* dpp, ceph::real_time* pmtime, optional_yield y)
@@ -3026,6 +3151,8 @@ int POSIXObject::delete_object(const DoutPrefixProvider* dpp,
     key.instance.clear();
     driver->get_bucket_cache()->remove_entry(dpp, b->get_name(), key);
   }
+  driver->get_quota_handler()->update_stats(b->get_owner(), b->get_key(),
+                                            -1, 0, state.accounted_size);
   return 0;
 }
 
@@ -4425,11 +4552,16 @@ int POSIXAtomicWriter::complete(size_t accounted_size, const std::string& etag,
                        uint32_t flags)
 {
   int ret;
+  uint64_t orig_size = 0;
+  auto exists = obj->check_exists(dpp);
+  if (exists) {
+    orig_size = obj->get_size();
+  }
 
   if (if_match) {
     if (strcmp(if_match, "*") == 0) {
       // test the object is existing
-      if (!obj->check_exists(dpp)) {
+      if (!exists) {
        return -ERR_PRECONDITION_FAILED;
       }
     } else {
@@ -4445,7 +4577,7 @@ int POSIXAtomicWriter::complete(size_t accounted_size, const std::string& etag,
   if (if_nomatch) {
     if (strcmp(if_nomatch, "*") == 0) {
       // test the object is not existing
-      if (obj->check_exists(dpp)) {
+      if (!exists) {
        return -ERR_PRECONDITION_FAILED;
       }
     } else {
@@ -4483,6 +4615,14 @@ int POSIXAtomicWriter::complete(size_t accounted_size, const std::string& etag,
     return ret;
   }
 
+  POSIXBucket *b = static_cast<POSIXBucket*>(obj->get_bucket());
+  if (!b) {
+      ldpp_dout(dpp, 0) << "ERROR: could not get bucket for " << obj->get_name() << dendl;
+      return -EINVAL;
+  }
+  driver->get_quota_handler()->update_stats(b->get_owner(), b->get_key(),
+                                            (exists ? 0 : 1), orig_size, accounted_size);
+
   return 0;
 }
 

diff --git a/src/rgw/driver/posix/rgw_sal_posix.h b/src/rgw/driver/posix/rgw_sal_posix.h
index 32e8c085a0f400c67457de3d064a5ae5529c06db..7779991ee827c6973298fa84bffd58ca88fa2583 100644 (file)
--- a/src/rgw/driver/posix/rgw_sal_posix.h
+++ b/src/rgw/driver/posix/rgw_sal_posix.h
@@ -17,6 +17,7 @@
 
 #include "rgw_sal_filter.h"
 #include "rgw_sal_store.h"
+#include "rgw_quota.h"
 #include <cstdint>
 #include <memory>
 #include "common/dout.h"
@@ -482,6 +483,7 @@ protected:
   std::unique_ptr<Directory> root_dir;
   int root_fd;
   RGWSyncModuleInstanceRef sync_module;
+  RGWQuotaHandler* quota_handler{nullptr};
 
 public:
   POSIXDriver(CephContext *_cct) : StoreDriver(), cct(_cct), zone(this)
@@ -708,9 +710,9 @@ public:
                             std::map<rgw_user_bucket, rgw_usage_log_entry>& usage) override { return 0; }
   virtual int trim_all_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch, optional_yield y) override { return 0; }
   virtual int get_config_key_val(std::string name, bufferlist* bl) override { return -ENOTSUP; }
-  virtual int meta_list_keys_init(const DoutPrefixProvider *dpp, const std::string& section, const std::string& marker, void** phandle) override { return 0; }
-  virtual int meta_list_keys_next(const DoutPrefixProvider *dpp, void* handle, int max, std::list<std::string>& keys, bool* truncated) override { return 0; }
-  virtual void meta_list_keys_complete(void* handle) override { return; }
+  virtual int meta_list_keys_init(const DoutPrefixProvider *dpp, const std::string& section, const std::string& marker, void** phandle) override;
+  virtual int meta_list_keys_next(const DoutPrefixProvider *dpp, void* handle, int max, std::list<std::string>& keys, bool* truncated) override;
+  virtual void meta_list_keys_complete(void* handle) override;
   virtual std::string meta_get_marker(void* handle) override { return ""; }
   virtual int meta_remove(const DoutPrefixProvider* dpp, std::string& metadata_key, optional_yield y) override { return 0; }
   virtual const RGWSyncModuleInstanceRef& get_sync_module() override { return sync_module; }
@@ -771,7 +773,7 @@ public:
   virtual const std::string& get_compression_type(const rgw_placement_rule& rule) override;
   virtual bool valid_placement(const rgw_placement_rule& rule) override { return true; } 
 
-  virtual void finalize(void) override {}
+  virtual void finalize(void) override;
 
   virtual CephContext* ctx(void) override { return userDB->ctx(); }
 
@@ -788,6 +790,8 @@ public:
    * by inotify or similar */
   int mint_listing_entry(
     const std::string& bucket, rgw_bucket_dir_entry& bde /* OUT */);
+
+  RGWQuotaHandler* get_quota_handler() {return quota_handler;}
 };
 
 class POSIXNotification : public StoreNotification {

diff --git a/src/test/rgw/test_rgw_posix_driver.cc b/src/test/rgw/test_rgw_posix_driver.cc
index 909bf4acee296150be080d2726bf6b857890485b..1b4574cc4c255a25f4d1080c17157b1ce9c249b9 100644 (file)
--- a/src/test/rgw/test_rgw_posix_driver.cc
+++ b/src/test/rgw/test_rgw_posix_driver.cc
@@ -1097,7 +1097,7 @@ public:
         }
       }
     }
-
+    quota_handler = RGWQuotaHandler::generate_handler(env->dpp, this, false);
     /* ordered listing cache */
     bucket_cache.reset(new BucketCache(
         this, base_path, cache_base, 100, 3, 3, 3));