Merge pull request ceph#9151 from fullerdj/wip-djf-15706

rbd: add methods to set and get snapshot limits

Reviewed-by: Jason Dillaman <[email protected]>

doc/man/8/rbd.rst

@@ -159,6 +159,10 @@ Parameters
    by examining the in-memory object map instead of querying RADOS for each
    object within the image.
 
+.. option:: --limit
+
+   Specifies the limit for the number of snapshots permitted.
+
 Commands
 ========
 
@@ -313,6 +317,13 @@ Commands
 
   This requires image format 2.
 
+:command:`snap limit set` [--limit] *limit* *image-spec*
+  Set a limit for the number of snapshots allowed on an image.
+
+:command:`snap limit clear` *image-spec*
+  Remove any previously set limit on the number of snapshots allowed on
+  an image.
+
 :command:`map` [-o | --options *map-options* ] [--read-only] *image-spec* | *snap-spec*
   Maps the specified image to a block device via the rbd kernel module.
 

src/cls/rbd/cls_rbd.cc

@@ -108,6 +108,8 @@ cls_method_handle_t h_metadata_set;
 cls_method_handle_t h_metadata_remove;
 cls_method_handle_t h_metadata_list;
 cls_method_handle_t h_metadata_get;
+cls_method_handle_t h_snapshot_get_limit;
+cls_method_handle_t h_snapshot_set_limit;
 cls_method_handle_t h_old_snapshots_list;
 cls_method_handle_t h_old_snapshot_add;
 cls_method_handle_t h_old_snapshot_remove;
@@ -1505,6 +1507,7 @@ int snapshot_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   bufferlist snap_namebl, snap_idbl;
   cls_rbd_snap snap_meta;
+  uint64_t snap_limit;
 
   try {
     bufferlist::iterator iter = in->begin();
@@ -1548,7 +1551,16 @@ int snapshot_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
     return r;
   }
 
+  r = read_key(hctx, "snap_limit", &snap_limit);
+  if (r == -ENOENT) {
+    snap_limit = UINT64_MAX;
+  } else if (r < 0) {
+    CLS_ERR("Could not read snapshot limit off disk: %s", cpp_strerror(r).c_str());
+    return r;
+  }
+
   int max_read = RBD_MAX_KEYS_READ;
+  uint64_t total_read = 0;
   string last_read = RBD_SNAP_KEY_PREFIX;
   do {
     map<string, bufferlist> vals;
@@ -1557,6 +1569,12 @@ int snapshot_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
     if (r < 0)
       return r;
 
+    total_read += vals.size();
+    if (total_read >= snap_limit) {
+      CLS_ERR("Attempt to create snapshot over limit of %lu", snap_limit);
+      return -EDQUOT;
+    }
+
     for (map<string, bufferlist>::iterator it = vals.begin();
 	 it != vals.end(); ++it) {
       cls_rbd_snap old_meta;
@@ -2681,6 +2699,50 @@ int metadata_get(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   return 0;
 }
 
+int snapshot_get_limit(cls_method_context_t hctx, bufferlist *in,
+		       bufferlist *out)
+{
+  int rc;
+  uint64_t snap_limit;
+
+  rc = read_key(hctx, "snap_limit", &snap_limit);
+  if (rc == -ENOENT) {
+    rc = 0;
+    ::encode(UINT64_MAX, *out);
+  } else {
+    ::encode(snap_limit, *out);
+  }
+
+  CLS_LOG(20, "read snapshot limit %lu", snap_limit);
+  return rc;
+}
+
+int snapshot_set_limit(cls_method_context_t hctx, bufferlist *in,
+		       bufferlist *out)
+{
+  int rc;
+  uint64_t new_limit;
+  bufferlist bl;
+
+  try {
+    bufferlist::iterator iter = in->begin();
+    ::decode(new_limit, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  if (new_limit == UINT64_MAX) {
+    CLS_LOG(20, "remove snapshot limit\n");
+    rc = cls_cxx_map_remove_key(hctx, "snap_limit");
+  } else {
+    CLS_LOG(20, "set snapshot limit to %lu\n", new_limit);
+    ::encode(new_limit, bl);
+    rc = cls_cxx_map_set_val(hctx, "snap_limit", &bl);
+  }
+
+  return rc;
+}
+
 
 /****************************** Old format *******************************/
 
@@ -2752,6 +2814,17 @@ int old_snapshot_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   if (header->snap_seq > snap_id)
     return -ESTALE;
 
+  uint64_t snap_limit;
+  rc = read_key(hctx, "snap_limit", &snap_limit);
+  if (rc == -ENOENT) {
+    snap_limit = UINT64_MAX;
+  } else if (rc < 0) {
+    return rc;
+  }
+
+  if (header->snap_count >= snap_limit)
+    return -EDQUOT;
+
   const char *cur_snap_name;
   for (cur_snap_name = snap_names; cur_snap_name < end; cur_snap_name += strlen(cur_snap_name) + 1) {
     if (strncmp(cur_snap_name, snap_name, end - cur_snap_name) == 0)
@@ -2973,6 +3046,7 @@ int old_snapshot_rename(cls_method_context_t hctx, bufferlist *in, bufferlist *o
   return 0;
 }
 
+
 namespace mirror {
 
 static const std::string UUID("mirror_uuid");
@@ -4430,6 +4504,12 @@ void __cls_init()
   cls_register_cxx_method(h_class, "metadata_get",
                           CLS_METHOD_RD,
 			  metadata_get, &h_metadata_get);
+  cls_register_cxx_method(h_class, "snapshot_get_limit",
+			  CLS_METHOD_RD,
+			  snapshot_get_limit, &h_snapshot_get_limit);
+  cls_register_cxx_method(h_class, "snapshot_set_limit",
+			  CLS_METHOD_WR,
+			  snapshot_set_limit, &h_snapshot_set_limit);
 
   /* methods for the rbd_children object */
   cls_register_cxx_method(h_class, "add_child",

src/cls/rbd/cls_rbd_client.cc

@@ -660,6 +660,33 @@ namespace librbd {
       op->exec("rbd", "set_protection_status", in);
     }
 
+    int snapshot_get_limit(librados::IoCtx *ioctx, const std::string &oid,
+			   uint64_t *limit)
+    {
+      bufferlist in, out;
+      int r =  ioctx->exec(oid, "rbd", "snapshot_get_limit", in, out);
+
+      if (r < 0) {
+	return r;
+      }
+
+      try {
+	bufferlist::iterator iter = out.begin();
+	::decode(*limit, iter);
+      } catch (const buffer::error &err) {
+	return -EBADMSG;
+      }
+
+      return 0;
+    }
+
+    void snapshot_set_limit(librados::ObjectWriteOperation *op, uint64_t limit)
+    {
+      bufferlist in;
+      ::encode(limit, in);
+      op->exec("rbd", "snapshot_set_limit", in);
+    }
+
     void get_stripe_unit_count_start(librados::ObjectReadOperation *op) {
       bufferlist empty_bl;
       op->exec("rbd", "get_stripe_unit_count", empty_bl);

src/cls/rbd/cls_rbd_client.h

@@ -124,6 +124,10 @@ namespace librbd {
 			      snapid_t snap_id, uint8_t protection_status);
     void set_protection_status(librados::ObjectWriteOperation *op,
                                snapid_t snap_id, uint8_t protection_status);
+    int snapshot_get_limit(librados::IoCtx *ioctx, const std::string &oid,
+			   uint64_t *limit);
+    void snapshot_set_limit(librados::ObjectWriteOperation *op,
+			    uint64_t limit);
 
     void get_stripe_unit_count_start(librados::ObjectReadOperation *op);
     int get_stripe_unit_count_finish(bufferlist::iterator *it,

src/include/rbd/librbd.h

@@ -354,6 +354,23 @@ CEPH_RBD_API int rbd_snap_unprotect(rbd_image_t image, const char *snap_name);
  */
 CEPH_RBD_API int rbd_snap_is_protected(rbd_image_t image, const char *snap_name,
 			               int *is_protected);
+/**
+ * Get the current snapshot limit for an image. If no limit is set,
+ * UINT64_MAX is returned.
+ *
+ * @param limit pointer where the limit will be stored on success
+ * @returns 0 on success, negative error code on failure
+ */
+CEPH_RBD_API int rbd_snap_get_limit(rbd_image_t image, uint64_t *limit);
+
+/**
+ * Set a limit for the number of snapshots that may be taken of an image.
+ *
+ * @param limit the maximum number of snapshots allowed in the future.
+ * @returns 0 on success, negative error code on failure
+ */
+CEPH_RBD_API int rbd_snap_set_limit(rbd_image_t image, uint64_t limit);
+
 CEPH_RBD_API int rbd_snap_set(rbd_image_t image, const char *snapname);
 
 CEPH_RBD_API int rbd_flatten(rbd_image_t image);

src/include/rbd/librbd.hpp

@@ -254,6 +254,8 @@ class CEPH_RBD_API Image
   int snap_is_protected(const char *snap_name, bool *is_protected);
   int snap_set(const char *snap_name);
   int snap_rename(const char *srcname, const char *dstname);
+  int snap_get_limit(uint64_t *limit);
+  int snap_set_limit(uint64_t limit);
 
   /* I/O */
   ssize_t read(uint64_t ofs, size_t len, ceph::bufferlist& bl);

src/librbd/CMakeLists.txt

@@ -60,6 +60,7 @@ set(librbd_internal_srcs
   operation/SnapshotRenameRequest.cc
   operation/SnapshotRollbackRequest.cc
   operation/SnapshotUnprotectRequest.cc
+  operation/SnapshotLimitRequest.cc
   operation/TrimRequest.cc)
 
 add_library(rbd_api STATIC librbd.cc)

src/librbd/Makefile.am

@@ -65,6 +65,7 @@ librbd_internal_la_SOURCES = \
 	librbd/operation/SnapshotRenameRequest.cc \
 	librbd/operation/SnapshotRollbackRequest.cc \
 	librbd/operation/SnapshotUnprotectRequest.cc \
+	librbd/operation/SnapshotLimitRequest.cc \
 	librbd/operation/TrimRequest.cc
 noinst_LTLIBRARIES += librbd_internal.la
 
@@ -156,6 +157,7 @@ noinst_HEADERS += \
 	librbd/operation/SnapshotRenameRequest.h \
 	librbd/operation/SnapshotRollbackRequest.h \
 	librbd/operation/SnapshotUnprotectRequest.h \
+	librbd/operation/SnapshotLimitRequest.h \
 	librbd/operation/TrimRequest.h
 
 endif # WITH_RBD

src/librbd/Operations.cc

@@ -22,6 +22,7 @@
 #include "librbd/operation/SnapshotRenameRequest.h"
 #include "librbd/operation/SnapshotRollbackRequest.h"
 #include "librbd/operation/SnapshotUnprotectRequest.h"
+#include "librbd/operation/SnapshotLimitRequest.h"
 #include <set>
 #include <boost/bind.hpp>
 
@@ -1087,6 +1088,56 @@ void Operations<I>::execute_snap_unprotect(const char *snap_name,
   request->send();
 }
 
+template <typename I>
+int Operations<I>::snap_set_limit(uint64_t limit) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": limit=" << limit << dendl;
+
+  if (m_image_ctx.read_only) {
+    return -EROFS;
+  }
+
+  int r = m_image_ctx.state->refresh_if_required();
+  if (r < 0) {
+    return r;
+  }
+
+  {
+    RWLock::RLocker owner_lock(m_image_ctx.owner_lock);
+    C_SaferCond limit_ctx;
+
+    if (m_image_ctx.exclusive_lock != nullptr &&
+	!m_image_ctx.exclusive_lock->is_lock_owner()) {
+      C_SaferCond lock_ctx;
+
+      m_image_ctx.exclusive_lock->request_lock(&lock_ctx);
+      r = lock_ctx.wait();
+      if (r < 0) {
+	return r;
+      }
+    }
+
+    execute_snap_set_limit(limit, &limit_ctx);
+    r = limit_ctx.wait();
+  }
+
+  return r;
+}
+
+template <typename I>
+void Operations<I>::execute_snap_set_limit(const uint64_t limit,
+					   Context *on_finish) {
+  assert(m_image_ctx.owner_lock.is_locked());
+
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": limit=" << limit
+                << dendl;
+
+  operation::SnapshotLimitRequest<I> *request =
+    new operation::SnapshotLimitRequest<I>(m_image_ctx, on_finish, limit);
+  request->send();
+}
+
 template <typename I>
 int Operations<I>::prepare_image_update() {
   assert(m_image_ctx.owner_lock.is_locked() &&

src/librbd/Operations.h

@@ -66,6 +66,9 @@ class Operations {
   int snap_unprotect(const char *snap_name);
   void execute_snap_unprotect(const char *snap_name, Context *on_finish);
 
+  int snap_set_limit(uint64_t limit);
+  void execute_snap_set_limit(uint64_t limit, Context *on_finish);
+
   int prepare_image_update();
 
 private:

src/librbd/internal.cc

@@ -2219,6 +2219,17 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force,
     return 0;
   }
 
+  int snap_get_limit(ImageCtx *ictx, uint64_t *limit)
+  {
+    return cls_client::snapshot_get_limit(&ictx->md_ctx, ictx->header_oid,
+					  limit);
+  }
+
+  int snap_set_limit(ImageCtx *ictx, uint64_t limit)
+  {
+    return ictx->operations->snap_set_limit(limit);
+  }
+
   struct CopyProgressCtx {
     explicit CopyProgressCtx(ProgressContext &p)
       : destictx(NULL), src_size(0), prog_ctx(p)

src/librbd/internal.h

@@ -138,6 +138,8 @@ namespace librbd {
 	     ProgressContext& prog_ctx, bool force=false);
   int snap_list(ImageCtx *ictx, std::vector<snap_info_t>& snaps);
   int snap_exists(ImageCtx *ictx, const char *snap_name, bool *exists);
+  int snap_get_limit(ImageCtx *ictx, uint64_t *limit);
+  int snap_set_limit(ImageCtx *ictx, uint64_t limit);
   int snap_is_protected(ImageCtx *ictx, const char *snap_name,
 			bool *is_protected);
   int copy(ImageCtx *ictx, IoCtx& dest_md_ctx, const char *destname,