Merge branch 'wip-rgw-quota'

Conflicts:
	src/test/cli/radosgw-admin/help.t

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>
Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

22 files changed, 875 insertions, 16 deletions

diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index 165ca437987..2851f2bd702 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -2,6 +2,7 @@
 
 #include "include/types.h"
 #include "cls/rgw/cls_rgw_ops.h"
+#include "cls/rgw/cls_rgw_client.h"
 #include "include/rados/librados.hpp"
 
 #include "common/debug.h"
@@ -157,6 +158,44 @@ int cls_rgw_get_dir_header(IoCtx& io_ctx, string& oid, rgw_bucket_dir_header *he
  return r;
 }
 
+class GetDirHeaderCompletion : public ObjectOperationCompletion {
+  RGWGetDirHeader_CB *ret_ctx;
+public:
+  GetDirHeaderCompletion(RGWGetDirHeader_CB *_ctx) : ret_ctx(_ctx) {}
+  ~GetDirHeaderCompletion() {
+    ret_ctx->put();
+  }
+  void handle_completion(int r, bufferlist& outbl) {
+    struct rgw_cls_list_ret ret;
+    try {
+      bufferlist::iterator iter = outbl.begin();
+      ::decode(ret, iter);
+    } catch (buffer::error& err) {
+      r = -EIO;
+    }
+
+    ret_ctx->handle_response(r, ret.dir.header);
+  };
+};
+
+int cls_rgw_get_dir_header_async(IoCtx& io_ctx, string& oid, RGWGetDirHeader_CB *ctx)
+{
+  bufferlist in, out;
+  struct rgw_cls_list_op call;
+  call.num_entries = 0;
+  ::encode(call, in);
+  ObjectReadOperation op;
+  GetDirHeaderCompletion *cb = new GetDirHeaderCompletion(ctx);
+  op.exec("rgw", "bucket_list", in, cb);
+  AioCompletion *c = librados::Rados::aio_create_completion(NULL, NULL, NULL);
+  int r = io_ctx.aio_operate(oid, c, &op, NULL);
+  c->release();
+  if (r < 0)
+    return r;
+
+  return 0;
+}
+
 int cls_rgw_bi_log_list(IoCtx& io_ctx, string& oid, string& marker, uint32_t max,
                     list<rgw_bi_log_entry>& entries, bool *truncated)
 {
diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 2ea5d9ca771..39bb3c9fc4a 100644
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -4,6 +4,13 @@
 #include "include/types.h"
 #include "include/rados/librados.hpp"
 #include "cls_rgw_types.h"
+#include "common/RefCountedObj.h"
+
+class RGWGetDirHeader_CB : public RefCountedObject {
+public:
+  virtual ~RGWGetDirHeader_CB() {}
+  virtual void handle_response(int r, rgw_bucket_dir_header& header) = 0;
+};
 
 /* bucket index */
 void cls_rgw_bucket_init(librados::ObjectWriteOperation& o);
@@ -27,6 +34,7 @@ int cls_rgw_bucket_check_index_op(librados::IoCtx& io_ctx, string& oid,
 int cls_rgw_bucket_rebuild_index_op(librados::IoCtx& io_ctx, string& oid);
   
 int cls_rgw_get_dir_header(librados::IoCtx& io_ctx, string& oid, rgw_bucket_dir_header *header);
+int cls_rgw_get_dir_header_async(librados::IoCtx& io_ctx, string& oid, RGWGetDirHeader_CB *ctx);
 
 void cls_rgw_encode_suggestion(char op, rgw_bucket_dir_entry& dirent, bufferlist& updates);
 
diff --git a/src/common/Formatter.h b/src/common/Formatter.h
index 27089ce04f2..ac68b7f461d 100644
--- a/src/common/Formatter.h
+++ b/src/common/Formatter.h
@@ -44,6 +44,9 @@ class Formatter {
   virtual void dump_int(const char *name, int64_t s) = 0;
   virtual void dump_float(const char *name, double d) = 0;
   virtual void dump_string(const char *name, std::string s) = 0;
+  virtual void dump_bool(const char *name, bool b) {
+    dump_format_unquoted(name, "%s", (b ? "true" : "false"));
+  }
   virtual std::ostream& dump_stream(const char *name) = 0;
   virtual void dump_format(const char *name, const char *fmt, ...) = 0;
   virtual void dump_format_unquoted(const char *name, const char *fmt, ...) = 0;
diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 2d3f981379b..700a210b412 100644
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -721,6 +721,10 @@ OPTION(rgw_data_log_num_shards, OPT_INT, 128) // number of objects to keep data
 OPTION(rgw_data_log_obj_prefix, OPT_STR, "data_log") // 
 OPTION(rgw_replica_log_obj_prefix, OPT_STR, "replica_log") // 
 
+OPTION(rgw_bucket_quota_ttl, OPT_INT, 600) // time for cached bucket stats to be cached within rgw instance
+OPTION(rgw_bucket_quota_soft_threshold, OPT_DOUBLE, 0.95) // threshold from which we don't rely on cached info for quota decisions
+OPTION(rgw_bucket_quota_cache_size, OPT_INT, 10000) // number of entries in bucket quota cache
+
 OPTION(mutex_perf_counter, OPT_BOOL, false) // enable/disable mutex perf counter
 
 // This will be set to true when it is safe to start threads.
diff --git a/src/common/lru_map.h b/src/common/lru_map.h
index 6e7f7b3786f..1e1acc95f76 100644
--- a/src/common/lru_map.h
+++ b/src/common/lru_map.h
@@ -21,41 +21,76 @@ class lru_map {
   size_t max;
 
 public:
+  class UpdateContext {
+    public:
+      virtual ~UpdateContext() {}
+
+      /* update should return true if object is updated */
+      virtual bool update(V *v) = 0;
+  };
+
+  bool _find(const K& key, V *value, UpdateContext *ctx);
+  void _add(const K& key, V& value);
+
+public:
   lru_map(int _max) : lock("lru_map"), max(_max) {}
   virtual ~lru_map() {}
 
   bool find(const K& key, V& value);
+
+  /*
+   * find_and_update()
+   *
+   * - will return true if object is found
+   * - if ctx is set will return true if object is found and updated
+   */
+  bool find_and_update(const K& key, V *value, UpdateContext *ctx);
   void add(const K& key, V& value);
   void erase(const K& key);
 };
 
 template <class K, class V>
-bool lru_map<K, V>::find(const K& key, V& value)
+bool lru_map<K, V>::_find(const K& key, V *value, UpdateContext *ctx)
 {
-  lock.Lock();
   typename std::map<K, entry>::iterator iter = entries.find(key);
   if (iter == entries.end()) {
-    lock.Unlock();
     return false;
   }
 
   entry& e = iter->second;
   entries_lru.erase(e.lru_iter);
 
-  value = e.value;
+  bool r = true;
+
+  if (ctx)
+    r = ctx->update(&e.value);
+
+  if (value)
+    *value = e.value;
 
   entries_lru.push_front(key);
   e.lru_iter = entries_lru.begin();
 
-  lock.Unlock();
+  return r;
+}
 
-  return true;
+template <class K, class V>
+bool lru_map<K, V>::find(const K& key, V& value)
+{
+  Mutex::Locker l(lock);
+  return _find(key, &value, NULL);
 }
 
 template <class K, class V>
-void lru_map<K, V>::add(const K& key, V& value)
+bool lru_map<K, V>::find_and_update(const K& key, V *value, UpdateContext *ctx)
+{
+  Mutex::Locker l(lock);
+  return _find(key, value, ctx);
+}
+
+template <class K, class V>
+void lru_map<K, V>::_add(const K& key, V& value)
 {
-  lock.Lock();
   typename std::map<K, entry>::iterator iter = entries.find(key);
   if (iter != entries.end()) {
     entry& e = iter->second;
@@ -74,8 +109,14 @@ void lru_map<K, V>::add(const K& key, V& value)
     entries.erase(iter);
     entries_lru.pop_back();
   }
-  
-  lock.Unlock();
+}
+
+
+template <class K, class V>
+void lru_map<K, V>::add(const K& key, V& value)
+{
+  Mutex::Locker l(lock);
+  _add(key, value);
 }
 
 template <class K, class V>
diff --git a/src/rgw/Makefile.am b/src/rgw/Makefile.am
index 24060b52e25..b92c35e08d6 100644
--- a/src/rgw/Makefile.am
+++ b/src/rgw/Makefile.am
@@ -31,7 +31,8 @@ librgw_la_SOURCES =  \
 	rgw/rgw_auth_s3.cc \
 	rgw/rgw_metadata.cc \
 	rgw/rgw_replica_log.cc \
-	rgw/rgw_keystone.cc
+	rgw/rgw_keystone.cc \
+	rgw/rgw_quota.cc
 librgw_la_CXXFLAGS = -Woverloaded-virtual ${AM_CXXFLAGS}
 noinst_LTLIBRARIES += librgw.la
 
@@ -124,6 +125,7 @@ noinst_HEADERS += \
 	rgw/rgw_http_client.h \
 	rgw/rgw_swift.h \
 	rgw/rgw_swift_auth.h \
+	rgw/rgw_quota.h \
 	rgw/rgw_rados.h \
 	rgw/rgw_replica_log.h \
 	rgw/rgw_resolve.h \
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 81abb231b6f..b23bf3ba5d4 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -62,6 +62,9 @@ void _usage()
   cerr << "  bucket check               check bucket index\n";
   cerr << "  object rm                  remove object\n";
   cerr << "  object unlink              unlink object from bucket index\n";
+  cerr << "  quota set                  set quota params\n";
+  cerr << "  quota enable               enable quota\n";
+  cerr << "  quota disable              disable quota\n";
   cerr << "  region get                 show region info\n";
   cerr << "  regions list               list all regions set on this cluster\n";
   cerr << "  region set                 set region info (requires infile)\n";
@@ -154,6 +157,11 @@ void _usage()
   cerr << "   --yes-i-really-mean-it    required for certain operations\n";
   cerr << "\n";
   cerr << "<date> := \"YYYY-MM-DD[ hh:mm:ss]\"\n";
+  cerr << "\nQuota options:\n";
+  cerr << "   --bucket                  specified bucket for quota command\n";
+  cerr << "   --max-objects             specify max objects\n";
+  cerr << "   --max-size                specify max size (in bytes)\n";
+  cerr << "   --quota-scope             scope of quota (bucket, user)\n";
   cerr << "\n";
   generic_client_usage();
 }
@@ -203,6 +211,9 @@ enum {
   OPT_OBJECT_RM,
   OPT_OBJECT_UNLINK,
   OPT_OBJECT_STAT,
+  OPT_QUOTA_SET,
+  OPT_QUOTA_ENABLE,
+  OPT_QUOTA_DISABLE,
   OPT_GC_LIST,
   OPT_GC_PROCESS,
   OPT_REGION_GET,
@@ -253,6 +264,7 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
       strcmp(cmd, "opstate") == 0 ||
       strcmp(cmd, "pool") == 0 ||
       strcmp(cmd, "pools") == 0 ||
+      strcmp(cmd, "quota") == 0 ||
       strcmp(cmd, "region") == 0 ||
       strcmp(cmd, "regions") == 0 ||
       strcmp(cmd, "region-map") == 0 ||
@@ -362,6 +374,13 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
       return OPT_REGION_SET;
     if (strcmp(cmd, "default") == 0)
       return OPT_REGION_DEFAULT;
+  } else if (strcmp(prev_cmd, "quota") == 0) {
+    if (strcmp(cmd, "set") == 0)
+      return OPT_QUOTA_SET;
+    if (strcmp(cmd, "enable") == 0)
+      return OPT_QUOTA_ENABLE;
+    if (strcmp(cmd, "disable") == 0)
+      return OPT_QUOTA_DISABLE;
   } else if (strcmp(prev_cmd, "regions") == 0) {
     if (strcmp(cmd, "list") == 0)
       return OPT_REGION_LIST;
@@ -660,6 +679,64 @@ static bool dump_string(const char *field_name, bufferlist& bl, Formatter *f)
   return true;
 }
 
+void set_quota_info(RGWQuotaInfo& quota, int opt_cmd, int64_t max_size, int64_t max_objects)
+{
+  switch (opt_cmd) {
+    case OPT_QUOTA_ENABLE:
+      quota.enabled = true;
+
+      // falling through on purpose
+
+    case OPT_QUOTA_SET:
+      if (max_objects >= 0) {
+        quota.max_objects = max_objects;
+      }
+      if (max_size >= 0) {
+        quota.max_size_kb = rgw_rounded_kb(max_size);
+      }
+      break;
+    case OPT_QUOTA_DISABLE:
+      quota.enabled = false;
+      break;
+  }
+}
+
+int set_bucket_quota(RGWRados *store, int opt_cmd, string& bucket_name, int64_t max_size, int64_t max_objects)
+{
+  RGWBucketInfo bucket_info;
+  map<string, bufferlist> attrs;
+  int r = store->get_bucket_info(NULL, bucket_name, bucket_info, NULL, &attrs);
+  if (r < 0) {
+    cerr << "could not get bucket info for bucket=" << bucket_name << ": " << cpp_strerror(-r) << std::endl;
+    return -r;
+  }
+
+  set_quota_info(bucket_info.quota, opt_cmd, max_size, max_objects);
+
+   r = store->put_bucket_instance_info(bucket_info, false, 0, &attrs);
+  if (r < 0) {
+    cerr << "ERROR: failed writing bucket instance info: " << cpp_strerror(-r) << std::endl;
+    return -r;
+  }
+  return 0;
+}
+
+int set_user_bucket_quota(int opt_cmd, RGWUser& user, RGWUserAdminOpState& op_state, int64_t max_size, int64_t max_objects)
+{
+  RGWUserInfo& user_info = op_state.get_user_info();
+
+  set_quota_info(user_info.bucket_quota, opt_cmd, max_size, max_objects);
+
+  op_state.set_bucket_quota(user_info.bucket_quota);
+
+  string err;
+  int r = user.modify(op_state, &err);
+  if (r < 0) {
+    cerr << "ERROR: failed updating user info: " << cpp_strerror(-r) << ": " << err << std::endl;
+    return -r;
+  }
+  return 0;
+}
 
 int main(int argc, char **argv) 
 {
@@ -721,6 +798,10 @@ int main(int argc, char **argv)
   string replica_log_type_str;
   ReplicaLogType replica_log_type = ReplicaLog_Invalid;
   string op_mask_str;
+  string quota_scope;
+
+  int64_t max_objects = -1;
+  int64_t max_size = -1;
 
   std::string val;
   std::ostringstream errs;
@@ -788,6 +869,10 @@ int main(int argc, char **argv)
       max_buckets = atoi(val.c_str());
     } else if (ceph_argparse_witharg(args, i, &val, "--max-entries", (char*)NULL)) {
       max_entries = atoi(val.c_str());
+    } else if (ceph_argparse_witharg(args, i, &val, "--max-size", (char*)NULL)) {
+      max_size = (int64_t)atoll(val.c_str());
+    } else if (ceph_argparse_witharg(args, i, &val, "--max-objects", (char*)NULL)) {
+      max_objects = (int64_t)atoll(val.c_str());
     } else if (ceph_argparse_witharg(args, i, &val, "--date", "--time", (char*)NULL)) {
       date = val;
       if (end_date.empty())
@@ -848,6 +933,8 @@ int main(int argc, char **argv)
       start_marker = val;
     } else if (ceph_argparse_witharg(args, i, &val, "--end-marker", (char*)NULL)) {
       end_marker = val;
+    } else if (ceph_argparse_witharg(args, i, &val, "--quota-scope", (char*)NULL)) {
+      quota_scope = val;
     } else if (ceph_argparse_witharg(args, i, &val, "--replica-log-type", (char*)NULL)) {
       replica_log_type_str = val;
       replica_log_type = get_replicalog_type(replica_log_type_str);
@@ -2228,5 +2315,28 @@ next:
         return -ret;
     }
   }
+
+  bool quota_op = (opt_cmd == OPT_QUOTA_SET || opt_cmd == OPT_QUOTA_ENABLE || opt_cmd == OPT_QUOTA_DISABLE);
+
+  if (quota_op) {
+    if (bucket_name.empty() && user_id.empty()) {
+      cerr << "ERROR: bucket name or uid is required for quota operation" << std::endl;
+      return EINVAL;
+    }
+
+    if (!bucket_name.empty()) {
+      if (!quota_scope.empty() && quota_scope != "bucket") {
+        cerr << "ERROR: invalid quota scope specification." << std::endl;
+        return EINVAL;
+      }
+      set_bucket_quota(store, opt_cmd, bucket_name, max_size, max_objects);
+    } else if (!user_id.empty()) {
+      if (quota_scope != "bucket") {
+        cerr << "ERROR: only bucket-level user quota can be handled. Please specify --quota-scope=bucket" << std::endl;
+        return EINVAL;
+      }
+      set_user_bucket_quota(opt_cmd, user, user_op, max_size, max_objects);
+    }
+  }
   return 0;
 }
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index 5356417f09a..3267bc51948 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -901,6 +901,7 @@ static int bucket_stats(RGWRados *store, std::string&  bucket_name, Formatter *f
   formatter->dump_int("mtime", mtime);
   formatter->dump_string("max_marker", max_marker);
   dump_bucket_usage(stats, formatter);
+  encode_json("bucket_quota", bucket_info.quota, formatter);
   formatter->close_section();
 
   return 0;
diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index 2c7c0c716be..baf60001a8b 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -29,6 +29,7 @@
 #include "include/utime.h"
 #include "rgw_acl.h"
 #include "rgw_cors.h"
+#include "rgw_quota.h"
 #include "cls/version/cls_version_types.h"
 #include "include/rados/librados.hpp"
 
@@ -90,6 +91,7 @@ using ceph::crypto::MD5;
 #define RGW_OP_TYPE_WRITE        0x02
 #define RGW_OP_TYPE_DELETE       0x04
 
+#define RGW_OP_TYPE_MODIFY       (RGW_OP_TYPE_WRITE | RGW_OP_TYPE_DELETE)
 #define RGW_OP_TYPE_ALL          (RGW_OP_TYPE_READ | RGW_OP_TYPE_WRITE | RGW_OP_TYPE_DELETE)
 
 #define RGW_DEFAULT_MAX_BUCKETS 1000
@@ -128,6 +130,7 @@ using ceph::crypto::MD5;
 #define ERR_NOT_FOUND            2023
 #define ERR_PERMANENT_REDIRECT   2024
 #define ERR_LOCKED               2025
+#define ERR_QUOTA_EXCEEDED       2026
 #define ERR_USER_SUSPENDED       2100
 #define ERR_INTERNAL_ERROR       2200
 
@@ -423,11 +426,12 @@ struct RGWUserInfo
   __u8 system;
   string default_placement;
   list<string> placement_tags;
+  RGWQuotaInfo bucket_quota;
 
   RGWUserInfo() : auid(0), suspended(0), max_buckets(RGW_DEFAULT_MAX_BUCKETS), op_mask(RGW_OP_TYPE_ALL), system(0) {}
 
   void encode(bufferlist& bl) const {
-     ENCODE_START(13, 9, bl);
+     ENCODE_START(14, 9, bl);
      ::encode(auid, bl);
      string access_key;
      string secret_key;
@@ -462,6 +466,7 @@ struct RGWUserInfo
      ::encode(system, bl);
      ::encode(default_placement, bl);
      ::encode(placement_tags, bl);
+     ::encode(bucket_quota, bl);
      ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator& bl) {
@@ -518,6 +523,9 @@ struct RGWUserInfo
       ::decode(default_placement, bl);
       ::decode(placement_tags, bl); /* tags of allowed placement rules */
     }
+    if (struct_v >= 14) {
+      ::decode(bucket_quota, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -599,6 +607,10 @@ struct rgw_bucket {
   void dump(Formatter *f) const;
   void decode_json(JSONObj *obj);
   static void generate_test_instances(list<rgw_bucket*>& o);
+
+  bool operator<(const rgw_bucket& b) const {
+    return name.compare(b.name) < 0;
+  }
 };
 WRITE_CLASS_ENCODER(rgw_bucket)
 
@@ -661,9 +673,10 @@ struct RGWBucketInfo
   bool has_instance_obj;
   RGWObjVersionTracker objv_tracker; /* we don't need to serialize this, for runtime tracking */
   obj_version ep_objv; /* entry point object version, for runtime tracking only */
+  RGWQuotaInfo quota;
 
   void encode(bufferlist& bl) const {
-     ENCODE_START(8, 4, bl);
+     ENCODE_START(9, 4, bl);
      ::encode(bucket, bl);
      ::encode(owner, bl);
      ::encode(flags, bl);
@@ -672,6 +685,7 @@ struct RGWBucketInfo
      ::encode(ct, bl);
      ::encode(placement_rule, bl);
      ::encode(has_instance_obj, bl);
+     ::encode(quota, bl);
      ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator& bl) {
@@ -692,6 +706,8 @@ struct RGWBucketInfo
        ::decode(placement_rule, bl);
      if (struct_v >= 8)
        ::decode(has_instance_obj, bl);
+     if (struct_v >= 9)
+       ::decode(quota, bl);
      DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -754,6 +770,8 @@ struct RGWBucketStats
   uint64_t num_kb;
   uint64_t num_kb_rounded;
   uint64_t num_objects;
+
+  RGWBucketStats() : num_kb(0), num_kb_rounded(0), num_objects(0) {}
 };
 
 struct req_state;
@@ -1213,6 +1231,11 @@ static inline const char *rgw_obj_category_name(RGWObjCategory category)
   return "unknown";
 }
 
+static inline uint64_t rgw_rounded_kb(uint64_t bytes)
+{
+  return (bytes + 1023) / 1024;
+}
+
 extern string rgw_string_unquote(const string& s);
 extern void parse_csv_string(const string& ival, vector<string>& ovals);
 extern int parse_key_value(string& in_str, string& key, string& val);
diff --git a/src/rgw/rgw_http_errors.h b/src/rgw/rgw_http_errors.h
index 6cb9fabf6c0..ba3e522651f 100644
--- a/src/rgw/rgw_http_errors.h
+++ b/src/rgw/rgw_http_errors.h
@@ -36,6 +36,7 @@ const static struct rgw_http_errors RGW_HTTP_ERRORS[] = {
     { EPERM, 403, "AccessDenied" },
     { ERR_USER_SUSPENDED, 403, "UserSuspended" },
     { ERR_REQUEST_TIME_SKEWED, 403, "RequestTimeTooSkewed" },
+    { ERR_QUOTA_EXCEEDED, 403, "QuotaExceeded" },
     { ENOENT, 404, "NoSuchKey" },
     { ERR_NO_SUCH_BUCKET, 404, "NoSuchBucket" },
     { ERR_NO_SUCH_UPLOAD, 404, "NoSuchUpload" },
diff --git a/src/rgw/rgw_json_enc.cc b/src/rgw/rgw_json_enc.cc
index 189e9ae961e..4d6b25374b9 100644
--- a/src/rgw/rgw_json_enc.cc
+++ b/src/rgw/rgw_json_enc.cc
@@ -396,6 +396,7 @@ void RGWUserInfo::dump(Formatter *f) const
   }
   encode_json("default_placement", default_placement, f);
   encode_json("placement_tags", placement_tags, f);
+  encode_json("bucket_quota", bucket_quota, f);
 }
 
 
@@ -446,6 +447,21 @@ void RGWUserInfo::decode_json(JSONObj *obj)
   system = (__u8)sys;
   JSONDecoder::decode_json("default_placement", default_placement, obj);
   JSONDecoder::decode_json("placement_tags", placement_tags, obj);
+  JSONDecoder::decode_json("bucket_quota", bucket_quota, obj);
+}
+
+void RGWQuotaInfo::dump(Formatter *f) const
+{
+  f->dump_bool("enabled", enabled);
+  f->dump_int("max_size_kb", max_size_kb);
+  f->dump_int("max_objects", max_objects);
+}
+
+void RGWQuotaInfo::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("max_size_kb", max_size_kb, obj);
+  JSONDecoder::decode_json("max_objects", max_objects, obj);
+  JSONDecoder::decode_json("enabled", enabled, obj);
 }
 
 void rgw_bucket::dump(Formatter *f) const
@@ -497,6 +513,7 @@ void RGWBucketInfo::dump(Formatter *f) const
   encode_json("region", region, f);
   encode_json("placement_rule", placement_rule, f);
   encode_json("has_instance_obj", has_instance_obj, f);
+  encode_json("quota", quota, f);
 }
 
 void RGWBucketInfo::decode_json(JSONObj *obj) {
@@ -507,6 +524,7 @@ void RGWBucketInfo::decode_json(JSONObj *obj) {
   JSONDecoder::decode_json("region", region, obj);
   JSONDecoder::decode_json("placement_rule", placement_rule, obj);
   JSONDecoder::decode_json("has_instance_obj", has_instance_obj, obj);
+  JSONDecoder::decode_json("quota", quota, obj);
 }
 
 void RGWObjEnt::dump(Formatter *f) const
@@ -673,12 +691,14 @@ void RGWRegionMap::dump(Formatter *f) const
 {
   encode_json("regions", regions, f);
   encode_json("master_region", master_region, f);
+  encode_json("bucket_quota", bucket_quota, f);
 }
 
 void RGWRegionMap::decode_json(JSONObj *obj)
 {
   JSONDecoder::decode_json("regions", regions, obj);
   JSONDecoder::decode_json("master_region", master_region, obj);
+  JSONDecoder::decode_json("bucket_quota", bucket_quota, obj);
 }
 
 void RGWMetadataLogInfo::dump(Formatter *f) const
diff --git a/src/rgw/rgw_main.cc b/src/rgw/rgw_main.cc
index 54db609521c..acaa5deffee 100644
--- a/src/rgw/rgw_main.cc
+++ b/src/rgw/rgw_main.cc
@@ -357,6 +357,13 @@ void RGWProcess::handle_request(RGWRequest *req)
     goto done;
   }
 
+  req->log(s, "init op");
+  ret = op->init_processing();
+  if (ret < 0) {
+    abort_early(s, op, ret);
+    goto done;
+  }
+
   req->log(s, "verifying op mask");
   ret = op->verify_op_mask();
   if (ret < 0) {
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index ca5ad3f2e7a..23f73e26531 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -1,7 +1,7 @@
 
 
-#include "rgw_metadata.h"
 #include "common/ceph_json.h"
+#include "rgw_metadata.h"
 #include "cls/version/cls_version_types.h"
 
 #include "rgw_rados.h"
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 114b8709a22..b9b4c53d696 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -421,6 +421,47 @@ int RGWOp::verify_op_mask()
   return 0;
 }
 
+int RGWOp::init_quota()
+{
+  /* no quota enforcement for system requests */
+  if (s->system_request)
+    return 0;
+
+  /* init quota related stuff */
+  if (!(s->user.op_mask & RGW_OP_TYPE_MODIFY)) {
+    return 0;
+  }
+
+  /* only interested in object related ops */
+  if (s->object_str.empty()) {
+    return 0;
+  }
+
+  if (s->bucket_info.quota.enabled) {
+    bucket_quota = s->bucket_info.quota;
+    return 0;
+  }
+  if (s->user.user_id == s->bucket_owner.get_id()) {
+    if (s->user.bucket_quota.enabled) {
+      bucket_quota = s->user.bucket_quota;
+      return 0;
+    }
+  } else {
+    RGWUserInfo owner_info;
+    int r = rgw_get_user_info_by_uid(store, s->bucket_info.owner, owner_info);
+    if (r < 0)
+      return r;
+
+    if (owner_info.bucket_quota.enabled) {
+      bucket_quota = owner_info.bucket_quota;
+      return 0;
+    }
+  }
+
+  bucket_quota = store->region_map.bucket_quota;
+  return 0;
+}
+
 static bool validate_cors_rule_method(RGWCORSRule *rule, const char *req_meth) {
   uint8_t flags = 0;
   if (strcmp(req_meth, "GET") == 0) flags = RGW_CORS_GET;
@@ -1363,6 +1404,14 @@ void RGWPutObj::execute()
     ldout(s->cct, 15) << "supplied_md5=" << supplied_md5 << dendl;
   }
 
+  if (!chunked_upload) { /* with chunked upload we don't know how big is the upload.
+                            we also check sizes at the end anyway */
+    ret = store->check_quota(s->bucket, bucket_quota, s->content_length);
+    if (ret < 0) {
+      goto done;
+    }
+  }
+
   if (supplied_etag) {
     strncpy(supplied_md5, supplied_etag, sizeof(supplied_md5) - 1);
     supplied_md5[sizeof(supplied_md5) - 1] = '\0';
@@ -1407,6 +1456,11 @@ void RGWPutObj::execute()
   s->obj_size = ofs;
   perfcounter->inc(l_rgw_put_b, s->obj_size);
 
+  ret = store->check_quota(s->bucket, bucket_quota, s->obj_size);
+  if (ret < 0) {
+    goto done;
+  }
+
   hash.Final(m);
 
   buf_to_hex(m, CEPH_CRYPTO_MD5_DIGESTSIZE, calc_md5);
diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index 948a11830c2..eee5ea99065 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -20,6 +20,7 @@
 #include "rgw_bucket.h"
 #include "rgw_acl.h"
 #include "rgw_cors.h"
+#include "rgw_quota.h"
 
 using namespace std;
 
@@ -36,10 +37,21 @@ protected:
   RGWRados *store;
   RGWCORSConfiguration bucket_cors;
   bool cors_exist;
+  RGWQuotaInfo bucket_quota;
+
+  virtual int init_quota();
 public:
   RGWOp() : s(NULL), dialect_handler(NULL), store(NULL), cors_exist(false) {}
   virtual ~RGWOp() {}
 
+  virtual int init_processing() {
+    int ret = init_quota();
+    if (ret < 0)
+      return ret;
+
+    return 0;
+  }
+
   virtual void init(RGWRados *store, struct req_state *s, RGWHandler *dialect_handler) {
     this->store = store;
     this->s = s;
diff --git a/src/rgw/rgw_quota.cc b/src/rgw/rgw_quota.cc
new file mode 100644
index 00000000000..66609ca723c
--- /dev/null
+++ b/src/rgw/rgw_quota.cc
@@ -0,0 +1,332 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2013 Inktank, Inc
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+
+#include "include/utime.h"
+#include "common/lru_map.h"
+#include "common/RefCountedObj.h"
+
+#include "rgw_common.h"
+#include "rgw_rados.h"
+#include "rgw_quota.h"
+
+#define dout_subsys ceph_subsys_rgw
+
+
+struct RGWQuotaBucketStats {
+  RGWBucketStats stats;
+  utime_t expiration;
+  utime_t async_refresh_time;
+};
+
+class RGWBucketStatsCache {
+  RGWRados *store;
+  lru_map<rgw_bucket, RGWQuotaBucketStats> stats_map;
+  RefCountedWaitObject *async_refcount;
+
+  int fetch_bucket_totals(rgw_bucket& bucket, RGWBucketStats& stats);
+
+public:
+  RGWBucketStatsCache(RGWRados *_store) : store(_store), stats_map(store->ctx()->_conf->rgw_bucket_quota_cache_size) {
+    async_refcount = new RefCountedWaitObject;
+  }
+  ~RGWBucketStatsCache() {
+    async_refcount->put_wait(); /* wait for all pending async requests to complete */
+  }
+
+  int get_bucket_stats(rgw_bucket& bucket, RGWBucketStats& stats, RGWQuotaInfo& quota);
+  void adjust_bucket_stats(rgw_bucket& bucket, int objs_delta, uint64_t added_bytes, uint64_t removed_bytes);
+
+  bool can_use_cached_stats(RGWQuotaInfo& quota, RGWBucketStats& stats);
+
+  void set_stats(rgw_bucket& bucket, RGWQuotaBucketStats& qs, RGWBucketStats& stats);
+  int async_refresh(rgw_bucket& bucket, RGWQuotaBucketStats& qs);
+  void async_refresh_response(rgw_bucket& bucket, RGWBucketStats& stats);
+};
+
+bool RGWBucketStatsCache::can_use_cached_stats(RGWQuotaInfo& quota, RGWBucketStats& cached_stats)
+{
+  if (quota.max_size_kb >= 0) {
+    if (quota.max_size_soft_threshold < 0) {
+      quota.max_size_soft_threshold = quota.max_size_kb * store->ctx()->_conf->rgw_bucket_quota_soft_threshold;
+    }
+
+    if (cached_stats.num_kb_rounded >= (uint64_t)quota.max_size_soft_threshold) {
+      ldout(store->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (size): "
+        << cached_stats.num_kb_rounded << " >= " << quota.max_size_soft_threshold << dendl;
+      return false;
+    }
+  }
+
+  if (quota.max_objects >= 0) {
+    if (quota.max_objs_soft_threshold < 0) {
+      quota.max_objs_soft_threshold = quota.max_objects * store->ctx()->_conf->rgw_bucket_quota_soft_threshold;
+    }
+
+    if (cached_stats.num_objects >= (uint64_t)quota.max_objs_soft_threshold) {
+      ldout(store->ctx(), 20) << "quota: can't use cached stats, exceeded soft threshold (num objs): "
+        << cached_stats.num_objects << " >= " << quota.max_objs_soft_threshold << dendl;
+      return false;
+    }
+  }
+
+  return true;
+}
+
+int RGWBucketStatsCache::fetch_bucket_totals(rgw_bucket& bucket, RGWBucketStats& stats)
+{
+  RGWBucketInfo bucket_info;
+
+  uint64_t bucket_ver;
+  uint64_t master_ver;
+
+  map<RGWObjCategory, RGWBucketStats> bucket_stats;
+  int r = store->get_bucket_stats(bucket, &bucket_ver, &master_ver, bucket_stats, NULL);
+  if (r < 0) {
+    ldout(store->ctx(), 0) << "could not get bucket info for bucket=" << bucket.name << dendl;
+    return r;
+  }
+
+  stats = RGWBucketStats();
+
+  map<RGWObjCategory, RGWBucketStats>::iterator iter;
+  for (iter = bucket_stats.begin(); iter != bucket_stats.end(); ++iter) {
+    RGWBucketStats& s = iter->second;
+    stats.num_kb += s.num_kb;
+    stats.num_kb_rounded += s.num_kb_rounded;
+    stats.num_objects += s.num_objects;
+  }
+
+  return 0;
+}
+
+class AsyncRefreshHandler : public RGWGetBucketStats_CB {
+  RGWRados *store;
+  RGWBucketStatsCache *cache;
+public:
+  AsyncRefreshHandler(RGWRados *_store, RGWBucketStatsCache *_cache, rgw_bucket& _bucket) : RGWGetBucketStats_CB(_bucket), store(_store), cache(_cache) {}
+
+  int init_fetch();
+
+  void handle_response(int r);
+};
+
+
+int AsyncRefreshHandler::init_fetch()
+{
+  ldout(store->ctx(), 20) << "initiating async quota refresh for bucket=" << bucket << dendl;
+  map<RGWObjCategory, RGWBucketStats> bucket_stats;
+  int r = store->get_bucket_stats_async(bucket, this);
+  if (r < 0) {
+    ldout(store->ctx(), 0) << "could not get bucket info for bucket=" << bucket.name << dendl;
+
+    /* get_bucket_stats_async() dropped our reference already */
+    return r;
+  }
+
+  return 0;
+}
+
+void AsyncRefreshHandler::handle_response(int r)
+{
+  if (r < 0) {
+    ldout(store->ctx(), 20) << "AsyncRefreshHandler::handle_response() r=" << r << dendl;
+    return; /* nothing to do here */
+  }
+
+  RGWBucketStats bs;
+
+  map<RGWObjCategory, RGWBucketStats>::iterator iter;
+  for (iter = stats->begin(); iter != stats->end(); ++iter) {
+    RGWBucketStats& s = iter->second;
+    bs.num_kb += s.num_kb;
+    bs.num_kb_rounded += s.num_kb_rounded;
+    bs.num_objects += s.num_objects;
+  }
+
+  cache->async_refresh_response(bucket, bs);
+}
+
+class RGWBucketStatsAsyncTestSet : public lru_map<rgw_bucket, RGWQuotaBucketStats>::UpdateContext {
+  int objs_delta;
+  uint64_t added_bytes;
+  uint64_t removed_bytes;
+public:
+  RGWBucketStatsAsyncTestSet() {}
+  bool update(RGWQuotaBucketStats *entry) {
+    if (entry->async_refresh_time.sec() == 0)
+      return false;
+
+    entry->async_refresh_time = utime_t(0, 0);
+
+    return true;
+  }
+};
+
+int RGWBucketStatsCache::async_refresh(rgw_bucket& bucket, RGWQuotaBucketStats& qs)
+{
+  /* protect against multiple updates */
+  RGWBucketStatsAsyncTestSet test_update;
+  if (!stats_map.find_and_update(bucket, NULL, &test_update)) {
+    /* most likely we just raced with another update */
+    return 0;
+  }
+
+  async_refcount->get();
+
+  AsyncRefreshHandler *handler = new AsyncRefreshHandler(store, this, bucket);
+
+  int ret = handler->init_fetch();
+  if (ret < 0) {
+    async_refcount->put();
+    handler->put();
+    return ret;
+  }
+
+  return 0;
+}
+
+void RGWBucketStatsCache::async_refresh_response(rgw_bucket& bucket, RGWBucketStats& stats)
+{
+  ldout(store->ctx(), 20) << "async stats refresh response for bucket=" << bucket << dendl;
+
+  RGWQuotaBucketStats qs;
+
+  stats_map.find(bucket, qs);
+
+  set_stats(bucket, qs, stats);
+
+  async_refcount->put();
+}
+
+void RGWBucketStatsCache::set_stats(rgw_bucket& bucket, RGWQuotaBucketStats& qs, RGWBucketStats& stats)
+{
+  qs.stats = stats;
+  qs.expiration = ceph_clock_now(store->ctx());
+  qs.async_refresh_time = qs.expiration;
+  qs.expiration += store->ctx()->_conf->rgw_bucket_quota_ttl;
+  qs.async_refresh_time += store->ctx()->_conf->rgw_bucket_quota_ttl / 2;
+
+  stats_map.add(bucket, qs);
+}
+
+int RGWBucketStatsCache::get_bucket_stats(rgw_bucket& bucket, RGWBucketStats& stats, RGWQuotaInfo& quota) {
+  RGWQuotaBucketStats qs;
+  utime_t now = ceph_clock_now(store->ctx());
+  if (stats_map.find(bucket, qs)) {
+    if (qs.async_refresh_time.sec() > 0 && now >= qs.async_refresh_time) {
+      int r = async_refresh(bucket, qs);
+      if (r < 0) {
+        ldout(store->ctx(), 0) << "ERROR: quota async refresh returned ret=" << r << dendl;
+
+        /* continue processing, might be a transient error, async refresh is just optimization */
+      }
+    }
+
+    if (can_use_cached_stats(quota, qs.stats) && qs.expiration > ceph_clock_now(store->ctx())) {
+      stats = qs.stats;
+      return 0;
+    }
+  }
+
+  int ret = fetch_bucket_totals(bucket, stats);
+  if (ret < 0 && ret != -ENOENT)
+    return ret;
+
+  set_stats(bucket, qs, stats);
+
+  return 0;
+}
+
+
+class RGWBucketStatsUpdate : public lru_map<rgw_bucket, RGWQuotaBucketStats>::UpdateContext {
+  int objs_delta;
+  uint64_t added_bytes;
+  uint64_t removed_bytes;
+public:
+  RGWBucketStatsUpdate(int _objs_delta, uint64_t _added_bytes, uint64_t _removed_bytes) : 
+                    objs_delta(_objs_delta), added_bytes(_added_bytes), removed_bytes(_removed_bytes) {}
+  bool update(RGWQuotaBucketStats *entry) {
+    uint64_t rounded_kb_added = rgw_rounded_kb(added_bytes);
+    uint64_t rounded_kb_removed = rgw_rounded_kb(removed_bytes);
+
+    entry->stats.num_kb_rounded += (rounded_kb_added - rounded_kb_removed);
+    entry->stats.num_kb += (added_bytes - removed_bytes) / 1024;
+    entry->stats.num_objects += objs_delta;
+
+    return true;
+  }
+};
+
+
+void RGWBucketStatsCache::adjust_bucket_stats(rgw_bucket& bucket, int objs_delta, uint64_t added_bytes, uint64_t removed_bytes)
+{
+  RGWBucketStatsUpdate update(objs_delta, added_bytes, removed_bytes);
+  stats_map.find_and_update(bucket, NULL, &update);
+}
+
+
+class RGWQuotaHandlerImpl : public RGWQuotaHandler {
+  RGWRados *store;
+  RGWBucketStatsCache stats_cache;
+public:
+  RGWQuotaHandlerImpl(RGWRados *_store) : store(_store), stats_cache(_store) {}
+  virtual int check_quota(rgw_bucket& bucket, RGWQuotaInfo& bucket_quota,
+			  uint64_t num_objs, uint64_t size) {
+    uint64_t size_kb = rgw_rounded_kb(size);
+    if (!bucket_quota.enabled) {
+      return 0;
+    }
+
+    RGWBucketStats stats;
+
+    int ret = stats_cache.get_bucket_stats(bucket, stats, bucket_quota);
+    if (ret < 0)
+      return ret;
+
+    ldout(store->ctx(), 20) << "bucket quota: max_objects=" << bucket_quota.max_objects
+                            << " max_size_kb=" << bucket_quota.max_size_kb << dendl;
+
+    if (bucket_quota.max_objects >= 0 &&
+        stats.num_objects + num_objs > (uint64_t)bucket_quota.max_objects) {
+      ldout(store->ctx(), 10) << "quota exceeded: stats.num_objects=" << stats.num_objects
+                              << " bucket_quota.max_objects=" << bucket_quota.max_objects << dendl;
+
+      return -ERR_QUOTA_EXCEEDED;
+    }
+    if (bucket_quota.max_size_kb >= 0 &&
+               stats.num_kb_rounded + size_kb > (uint64_t)bucket_quota.max_size_kb) {
+      ldout(store->ctx(), 10) << "quota exceeded: stats.num_kb_rounded=" << stats.num_kb_rounded << " size_kb=" << size_kb
+                              << " bucket_quota.max_size_kb=" << bucket_quota.max_size_kb << dendl;
+      return -ERR_QUOTA_EXCEEDED;
+    }
+
+    return 0;
+  }
+
+  virtual void update_stats(rgw_bucket& bucket, int obj_delta, uint64_t added_bytes, uint64_t removed_bytes) {
+    stats_cache.adjust_bucket_stats(bucket, obj_delta, added_bytes, removed_bytes);
+  };
+};
+
+
+RGWQuotaHandler *RGWQuotaHandler::generate_handler(RGWRados *store)
+{
+  return new RGWQuotaHandlerImpl(store);
+};
+
+void RGWQuotaHandler::free_handler(RGWQuotaHandler *handler)
+{
+  delete handler;
+}
diff --git a/src/rgw/rgw_quota.h b/src/rgw/rgw_quota.h
new file mode 100644
index 00000000000..2f8f28e85a2
--- /dev/null
+++ b/src/rgw/rgw_quota.h
@@ -0,0 +1,74 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2013 Inktank, Inc
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef CEPH_RGW_QUOTA_H
+#define CEPH_RGW_QUOTA_H
+
+
+#include "include/utime.h"
+#include "include/atomic.h"
+#include "common/lru_map.h"
+
+class RGWRados;
+class JSONObj;
+
+struct RGWQuotaInfo {
+  int64_t max_size_kb;
+  int64_t max_objects;
+  bool enabled;
+  int64_t max_size_soft_threshold;
+  int64_t max_objs_soft_threshold;
+
+  RGWQuotaInfo() : max_size_kb(-1), max_objects(-1), enabled(false),
+                   max_size_soft_threshold(-1), max_objs_soft_threshold(-1) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(max_size_kb, bl);
+    ::encode(max_objects, bl);
+    ::encode(enabled, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(max_size_kb, bl);
+    ::decode(max_objects, bl);
+    ::decode(enabled, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(Formatter *f) const;
+
+  void decode_json(JSONObj *obj);
+
+};
+WRITE_CLASS_ENCODER(RGWQuotaInfo)
+
+class rgw_bucket;
+
+class RGWQuotaHandler {
+public:
+  RGWQuotaHandler() {}
+  virtual ~RGWQuotaHandler() {
+  }
+  virtual int check_quota(rgw_bucket& bucket, RGWQuotaInfo& bucket_quota,
+			  uint64_t num_objs, uint64_t size) = 0;
+
+  virtual void update_stats(rgw_bucket& bucket, int obj_delta, uint64_t added_bytes, uint64_t removed_bytes) = 0;
+
+  static RGWQuotaHandler *generate_handler(RGWRados *store);
+  static void free_handler(RGWQuotaHandler *handler);
+};
+
+#endif
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 8b4d18f4e68..9f0a900f3d3 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -357,16 +357,20 @@ int RGWZoneParams::store_info(CephContext *cct, RGWRados *store, RGWRegion& regi
 }
 
 void RGWRegionMap::encode(bufferlist& bl) const {
-  ENCODE_START(1, 1, bl);
+  ENCODE_START(2, 1, bl);
   ::encode(regions, bl);
   ::encode(master_region, bl);
+  ::encode(bucket_quota, bl);
   ENCODE_FINISH(bl);
 }
 
 void RGWRegionMap::decode(bufferlist::iterator& bl) {
-  DECODE_START(1, bl);
+  DECODE_START(2, bl);
   ::decode(regions, bl);
   ::decode(master_region, bl);
+
+  if (struct_v >= 2)
+    ::decode(bucket_quota, bl);
   DECODE_FINISH(bl);
 
   regions_by_api.clear();
@@ -851,6 +855,7 @@ void RGWRados::finalize()
     RGWRESTConn *conn = iter->second;
     delete conn;
   }
+  RGWQuotaHandler::free_handler(quota_handler);
 }
 
 /** 
@@ -962,6 +967,8 @@ int RGWRados::init_complete()
   if (use_gc_thread)
     gc->start_processor();
 
+  quota_handler = RGWQuotaHandler::generate_handler(this);
+
   return ret;
 }
 
@@ -2342,6 +2349,11 @@ int RGWRados::put_obj_meta_impl(void *ctx, rgw_obj& obj,  uint64_t size,
     *mtime = set_mtime;
   }
 
+  if (state) {
+    /* update quota cache */
+    quota_handler->update_stats(bucket, (state->exists ? 0 : 1), size, state->size);
+  }
+
   return 0;
 
 done_cancel:
@@ -3211,6 +3223,11 @@ int RGWRados::delete_obj_impl(void *ctx, rgw_obj& obj, RGWObjVersionTracker *obj
   if (ret_not_existed)
     return -ENOENT;
 
+  if (state) {
+    /* update quota cache */
+    quota_handler->update_stats(bucket, -1, 0, state->size);
+  }
+
   return 0;
 }
 
@@ -4598,6 +4615,38 @@ int RGWRados::get_bucket_stats(rgw_bucket& bucket, uint64_t *bucket_ver, uint64_
   return 0;
 }
 
+class RGWGetBucketStatsContext : public RGWGetDirHeader_CB {
+  RGWGetBucketStats_CB *cb;
+
+public:
+  RGWGetBucketStatsContext(RGWGetBucketStats_CB *_cb) : cb(_cb) {}
+  void handle_response(int r, rgw_bucket_dir_header& header) {
+    map<RGWObjCategory, RGWBucketStats> stats;
+
+    if (r >= 0) {
+      translate_raw_stats(header, stats);
+      cb->set_response(header.ver, header.master_ver, &stats, header.max_marker);
+    }
+
+    cb->handle_response(r);
+
+    cb->put();
+  }
+};
+
+int RGWRados::get_bucket_stats_async(rgw_bucket& bucket, RGWGetBucketStats_CB *ctx)
+{
+  RGWGetBucketStatsContext *get_ctx = new RGWGetBucketStatsContext(ctx);
+  int r = cls_bucket_head_async(bucket, get_ctx);
+  if (r < 0) {
+    ctx->put();
+    delete get_ctx;
+    return r;
+  }
+
+  return 0;
+}
+
 void RGWRados::get_bucket_instance_entry(rgw_bucket& bucket, string& entry)
 {
   entry = bucket.name + ":" + bucket.bucket_id;
@@ -5480,6 +5529,25 @@ int RGWRados::cls_bucket_head(rgw_bucket& bucket, struct rgw_bucket_dir_header&
   return 0;
 }
 
+int RGWRados::cls_bucket_head_async(rgw_bucket& bucket, RGWGetDirHeader_CB *ctx)
+{
+  librados::IoCtx index_ctx;
+  string oid;
+  int r = open_bucket_index(bucket, index_ctx, oid);
+  if (r < 0)
+    return r;
+
+  r = cls_rgw_get_dir_header_async(index_ctx, oid, ctx);
+  if (r < 0)
+    return r;
+
+  return 0;
+}
+
+int RGWRados::check_quota(rgw_bucket& bucket, RGWQuotaInfo& quota_info, uint64_t obj_size)
+{
+  return quota_handler->check_quota(bucket, quota_info, 1, obj_size);
+}
 
 class IntentLogNameFilter : public RGWAccessListFilter
 {
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 65765c414aa..52b898123d4 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -636,6 +636,8 @@ struct RGWRegionMap {
 
   string master_region;
 
+  RGWQuotaInfo bucket_quota;
+
   RGWRegionMap() : lock("RGWRegionMap") {}
 
   void encode(bufferlist& bl) const;
@@ -759,6 +761,29 @@ public:
   int renew_state();
 };
 
+class RGWGetBucketStats_CB : public RefCountedObject {
+protected:
+  rgw_bucket bucket;
+  uint64_t bucket_ver;
+  uint64_t master_ver;
+  map<RGWObjCategory, RGWBucketStats> *stats;
+  string max_marker;
+public:
+  RGWGetBucketStats_CB(rgw_bucket& _bucket) : bucket(_bucket), stats(NULL) {}
+  virtual ~RGWGetBucketStats_CB() {}
+  virtual void handle_response(int r) = 0;
+  virtual void set_response(uint64_t _bucket_ver, uint64_t _master_ver,
+                            map<RGWObjCategory, RGWBucketStats> *_stats,
+                            const string &_max_marker) {
+    bucket_ver = _bucket_ver;
+    master_ver = _master_ver;
+    stats = _stats;
+    max_marker = _max_marker;
+  }
+};
+
+class RGWGetDirHeader_CB;
+
 
 class RGWRados
 {
@@ -862,6 +887,8 @@ protected:
   string region_name;
   string zone_name;
 
+  RGWQuotaHandler *quota_handler;
+
 public:
   RGWRados() : lock("rados_timer_lock"), timer(NULL),
                gc(NULL), use_gc_thread(false),
@@ -870,6 +897,7 @@ public:
                bucket_id_lock("rados_bucket_id"), max_bucket_id(0),
                cct(NULL), rados(NULL),
                pools_initialized(false),
+               quota_handler(NULL),
                rest_master_conn(NULL),
                meta_mgr(NULL), data_log(NULL) {}
 
@@ -1290,6 +1318,7 @@ public:
   int decode_policy(bufferlist& bl, ACLOwner *owner);
   int get_bucket_stats(rgw_bucket& bucket, uint64_t *bucket_ver, uint64_t *master_ver, map<RGWObjCategory, RGWBucketStats>& stats,
                        string *max_marker);
+  int get_bucket_stats_async(rgw_bucket& bucket, RGWGetBucketStats_CB *cb);
   void get_bucket_instance_obj(rgw_bucket& bucket, rgw_obj& obj);
   void get_bucket_instance_entry(rgw_bucket& bucket, string& entry);
   void get_bucket_meta_oid(rgw_bucket& bucket, string& oid);
@@ -1321,6 +1350,7 @@ public:
                       map<string, RGWObjEnt>& m, bool *is_truncated,
                       string *last_entry, bool (*force_check_filter)(const string&  name) = NULL);
   int cls_bucket_head(rgw_bucket& bucket, struct rgw_bucket_dir_header& header);
+  int cls_bucket_head_async(rgw_bucket& bucket, RGWGetDirHeader_CB *ctx);
   int prepare_update_index(RGWObjState *state, rgw_bucket& bucket,
                            RGWModifyOp op, rgw_obj& oid, string& tag);
   int complete_update_index(rgw_bucket& bucket, string& oid, string& tag, int64_t poolid, uint64_t epoch, uint64_t size,
@@ -1376,6 +1406,8 @@ public:
   int bucket_rebuild_index(rgw_bucket& bucket);
   int remove_objs_from_index(rgw_bucket& bucket, list<string>& oid_list);
 
+  int check_quota(rgw_bucket& bucket, RGWQuotaInfo& quota_info, uint64_t obj_size);
+
   string unique_id(uint64_t unique_num) {
     char buf[32];
     snprintf(buf, sizeof(buf), ".%llu.%llu", (unsigned long long)instance_id(), (unsigned long long)unique_num);
diff --git a/src/rgw/rgw_user.cc b/src/rgw/rgw_user.cc
index 5e5b5c564bb..dc529e3d48d 100644
--- a/src/rgw/rgw_user.cc
+++ b/src/rgw/rgw_user.cc
@@ -1682,6 +1682,9 @@ int RGWUser::execute_add(RGWUserAdminOpState& op_state, std::string *err_msg)
   if (op_state.op_mask_specified)
     user_info.op_mask = op_state.get_op_mask();
 
+  if (op_state.has_bucket_quota())
+    user_info.bucket_quota = op_state.get_bucket_quota();
+
   // update the request
   op_state.set_user_info(user_info);
   op_state.set_populated();
@@ -1884,6 +1887,9 @@ int RGWUser::execute_modify(RGWUserAdminOpState& op_state, std::string *err_msg)
   if (op_state.op_mask_specified)
     user_info.op_mask = op_state.get_op_mask();
 
+  if (op_state.has_bucket_quota())
+    user_info.bucket_quota = op_state.get_bucket_quota();
+
   if (op_state.has_suspension_op()) {
     __u8 suspended = op_state.get_suspension_status();
     user_info.suspended = suspended;
diff --git a/src/rgw/rgw_user.h b/src/rgw/rgw_user.h
index 32bcf199001..e71b8f81778 100644
--- a/src/rgw/rgw_user.h
+++ b/src/rgw/rgw_user.h
@@ -172,6 +172,10 @@ struct RGWUserAdminOpState {
   bool subuser_params_checked;
   bool user_params_checked;
 
+  bool bucket_quota_specified;
+
+  RGWQuotaInfo bucket_quota;
+
   void set_access_key(std::string& access_key) {
     if (access_key.empty())
       return;
@@ -285,6 +289,12 @@ struct RGWUserAdminOpState {
     key_op = true;
   }
 
+  void set_bucket_quota(RGWQuotaInfo& quota)
+  {
+    bucket_quota = quota;
+    bucket_quota_specified = true;
+  }
+
   bool is_populated() { return populated; };
   bool is_initialized() { return initialized; };
   bool has_existing_user() { return existing_user; };
@@ -303,6 +313,7 @@ struct RGWUserAdminOpState {
   bool will_purge_keys() { return purge_keys; };
   bool will_purge_data() { return purge_data; };
   bool will_generate_subuser() { return gen_subuser; };
+  bool has_bucket_quota() { return bucket_quota_specified; }
   void set_populated() { populated = true; };
   void clear_populated() { populated = false; };
   void set_initialized() { initialized = true; };
@@ -317,6 +328,7 @@ struct RGWUserAdminOpState {
   uint32_t get_subuser_perm() { return perm_mask; };
   uint32_t get_max_buckets() { return max_buckets; };
   uint32_t get_op_mask() { return op_mask; };
+  RGWQuotaInfo& get_bucket_quota() { return bucket_quota; }
 
   std::string get_user_id() { return user_id; };
   std::string get_subuser() { return subuser; };
@@ -403,6 +415,7 @@ struct RGWUserAdminOpState {
     key_params_checked = false;
     subuser_params_checked = false;
     user_params_checked = false;
+    bucket_quota_specified = false;
   }
 };
 
diff --git a/src/test/cli/radosgw-admin/help.t b/src/test/cli/radosgw-admin/help.t
index 2def60107dc..4fe30b1cda7 100644
--- a/src/test/cli/radosgw-admin/help.t
+++ b/src/test/cli/radosgw-admin/help.t
@@ -23,6 +23,9 @@
     bucket check               check bucket index
     object rm                  remove object
     object unlink              unlink object from bucket index
+    quota set                  set quota params
+    quota enable               enable quota
+    quota disable              disable quota
     region get                 show region info
     regions list               list all regions set on this cluster
     region set                 set region info (requires infile)
@@ -116,6 +119,12 @@
   
   <date> := "YYYY-MM-DD[ hh:mm:ss]"
   
+  Quota options:
+     --bucket                  specified bucket for quota command
+     --max-objects             specify max objects
+     --max-size                specify max size (in bytes)
+     --quota-scope             scope of quota (bucket, user)
+  
     --conf/-c FILE    read configuration from the given configuration file
     --id/-i ID        set ID portion of my name
     --name/-n TYPE.ID set name