librbd: track new operation features within image

This will initially be utilized to restrict older clients from
performing operations against an image if (1) it doesn't
support the new feature bit, or (2) doesn't support the specific
enabled op feature.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

src/cls/rbd/cls_rbd.cc

@@ -350,6 +350,12 @@ int set_features(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
     return r;
   }
 
+  if ((mask & RBD_FEATURES_INTERNAL) != 0ULL) {
+    CLS_ERR("Attempting to set internal feature: %" PRIu64,
+            static_cast<uint64_t>(mask & RBD_FEATURES_INTERNAL));
+    return -EINVAL;
+  }
+
   // newer clients might attempt to mask off features we don't support
   mask &= RBD_FEATURES_ALL;
 
@@ -920,6 +926,112 @@ int set_flags(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   return 0;
 }
 
+/**
+ * Get the operation-based image features
+ *
+ * Input:
+ *
+ * Output:
+ * @param bitmask of enabled op features (uint64_t)
+ * @returns 0 on success, negative error code on failure
+ */
+int op_features_get(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  CLS_LOG(20, "op_features_get");
+
+  uint64_t op_features = 0;
+  int r = read_key(hctx, "op_features", &op_features);
+  if (r < 0 && r != -ENOENT) {
+    CLS_ERR("failed to read op features off disk: %s", cpp_strerror(r).c_str());
+    return r;
+  }
+
+  encode(op_features, *out);
+  return 0;
+}
+
+/**
+ * Set the operation-based image features
+ *
+ * Input:
+ * @param op_features image op features
+ * @param mask image op feature mask
+ *
+ * Output:
+ * none
+ *
+ * @returns 0 on success, negative error code upon failure
+ */
+int op_features_set(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  uint64_t op_features;
+  uint64_t mask;
+  bufferlist::iterator iter = in->begin();
+  try {
+    decode(op_features, iter);
+    decode(mask, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  uint64_t unsupported_op_features = (mask & ~RBD_OPERATION_FEATURES_ALL);
+  if (unsupported_op_features != 0ULL) {
+    CLS_ERR("unsupported op features: %" PRIu64, unsupported_op_features);
+    return -EINVAL;
+  }
+
+  uint64_t orig_features;
+  int r = read_key(hctx, "features", &orig_features);
+  if (r < 0) {
+    CLS_ERR("failed to read features off disk: %s", cpp_strerror(r).c_str());
+    return r;
+  }
+
+  uint64_t orig_op_features = 0;
+  r = read_key(hctx, "op_features", &orig_op_features);
+  if (r < 0 && r != -ENOENT) {
+    CLS_ERR("Could not read op features off disk: %s", cpp_strerror(r).c_str());
+    return r;
+  }
+
+  op_features = (orig_op_features & ~mask) | (op_features & mask);
+  CLS_LOG(10, "set_features op_features=%" PRIu64 " orig_op_features=%" PRIu64,
+          op_features, orig_op_features);
+
+  uint64_t features = orig_features;
+  if (op_features == 0ULL) {
+    features &= ~RBD_FEATURE_OPERATIONS;
+
+    r = cls_cxx_map_remove_key(hctx, "op_features");
+    if (r == -ENOENT) {
+      r = 0;
+    }
+  } else {
+    features |= RBD_FEATURE_OPERATIONS;
+
+    bufferlist bl;
+    encode(op_features, bl);
+    r = cls_cxx_map_set_val(hctx, "op_features", &bl);
+  }
+
+  if (r < 0) {
+    CLS_ERR("error updating op features: %s", cpp_strerror(r).c_str());
+    return r;
+  }
+
+  if (features != orig_features) {
+    bufferlist bl;
+    encode(features, bl);
+    r = cls_cxx_map_set_val(hctx, "features", &bl);
+    if (r < 0) {
+      CLS_ERR("error updating features: %s", cpp_strerror(r).c_str());
+      return r;
+    }
+  }
+
+  return 0;
+}
+
 /**
  * get the current parent, if any
  *
@@ -5536,6 +5648,8 @@ CLS_INIT(rbd)
   cls_method_handle_t h_get_create_timestamp;
   cls_method_handle_t h_get_flags;
   cls_method_handle_t h_set_flags;
+  cls_method_handle_t h_op_features_get;
+  cls_method_handle_t h_op_features_set;
   cls_method_handle_t h_remove_parent;
   cls_method_handle_t h_add_child;
   cls_method_handle_t h_remove_child;
@@ -5697,6 +5811,11 @@ CLS_INIT(rbd)
   cls_register_cxx_method(h_class, "set_flags",
                           CLS_METHOD_RD | CLS_METHOD_WR,
                           set_flags, &h_set_flags);
+  cls_register_cxx_method(h_class, "op_features_get", CLS_METHOD_RD,
+                          op_features_get, &h_op_features_get);
+  cls_register_cxx_method(h_class, "op_features_set",
+                          CLS_METHOD_RD | CLS_METHOD_WR,
+                          op_features_set, &h_op_features_set);
   cls_register_cxx_method(h_class, "metadata_list",
                           CLS_METHOD_RD,
 			  metadata_list, &h_metadata_list);

src/cls/rbd/cls_rbd_client.cc

@@ -403,6 +403,56 @@ namespace librbd {
       op->exec("rbd", "set_flags", inbl);
     }
 
+    void op_features_get_start(librados::ObjectReadOperation *op)
+    {
+      bufferlist in_bl;
+      op->exec("rbd", "op_features_get", in_bl);
+    }
+
+    int op_features_get_finish(bufferlist::iterator *it, uint64_t *op_features)
+    {
+      try {
+        decode(*op_features, *it);
+      } catch (const buffer::error &err) {
+        return -EBADMSG;
+      }
+      return 0;
+    }
+
+    int op_features_get(librados::IoCtx *ioctx, const std::string &oid,
+		        uint64_t *op_features)
+    {
+      librados::ObjectReadOperation op;
+      op_features_get_start(&op);
+
+      bufferlist out_bl;
+      int r = ioctx->operate(oid, &op, &out_bl);
+      if (r < 0) {
+        return r;
+      }
+
+      bufferlist::iterator it = out_bl.begin();
+      return op_features_get_finish(&it, op_features);
+    }
+
+    void op_features_set(librados::ObjectWriteOperation *op,
+                         uint64_t op_features, uint64_t mask)
+    {
+      bufferlist inbl;
+      encode(op_features, inbl);
+      encode(mask, inbl);
+      op->exec("rbd", "op_features_set", inbl);
+    }
+
+    int op_features_set(librados::IoCtx *ioctx, const std::string &oid,
+                        uint64_t op_features, uint64_t mask)
+    {
+      librados::ObjectWriteOperation op;
+      op_features_set(&op, op_features, mask);
+
+      return ioctx->operate(oid, &op);
+    }
+
     int remove_parent(librados::IoCtx *ioctx, const std::string &oid)
     {
       librados::ObjectWriteOperation op;

src/cls/rbd/cls_rbd_client.h

@@ -89,6 +89,14 @@ namespace librbd {
 		  vector<uint64_t> *snap_flags);
     void set_flags(librados::ObjectWriteOperation *op, snapid_t snap_id,
                    uint64_t flags, uint64_t mask);
+    void op_features_get_start(librados::ObjectReadOperation *op);
+    int op_features_get_finish(bufferlist::iterator *it, uint64_t *op_features);
+    int op_features_get(librados::IoCtx *ioctx, const std::string &oid,
+		        uint64_t *op_features);
+    void op_features_set(librados::ObjectWriteOperation *op,
+                         uint64_t op_features, uint64_t mask);
+    int op_features_set(librados::IoCtx *ioctx, const std::string &oid,
+                        uint64_t op_features, uint64_t mask);
     int remove_parent(librados::IoCtx *ioctx, const std::string &oid);
     void remove_parent(librados::ObjectWriteOperation *op);
     int add_child(librados::IoCtx *ioctx, const std::string &oid,

src/common/options.cc

@@ -5732,7 +5732,7 @@ static std::vector<Option> get_rbd_options() {
         {RBD_FEATURE_NAME_JOURNALING, RBD_FEATURE_JOURNALING},
         {RBD_FEATURE_NAME_DATA_POOL, RBD_FEATURE_DATA_POOL},
       };
-      static_assert((RBD_FEATURE_DATA_POOL << 1) > RBD_FEATURES_ALL,
+      static_assert((RBD_FEATURE_OPERATIONS << 1) > RBD_FEATURES_ALL,
                     "new RBD feature added");
 
       // convert user-friendly comma delimited feature name list to a bitmask
@@ -5752,6 +5752,15 @@ static std::vector<Option> get_rbd_options() {
              << std::hex << unsupported_features;
           *error_message = ss.str();
         }
+        uint64_t internal_features = (features & RBD_FEATURES_INTERNAL);
+        if (internal_features != 0ULL) {
+          features &= ~RBD_FEATURES_INTERNAL;
+
+          std::stringstream ss;
+          ss << "ignoring internal feature mask 0x"
+             << std::hex << internal_features;
+          *error_message = ss.str();
+        }
       } catch (const boost::bad_lexical_cast& ) {
         int r = 0;
         std::vector<std::string> feature_names;

src/include/rbd/features.h

@@ -9,6 +9,7 @@
 #define RBD_FEATURE_DEEP_FLATTEN        (1ULL<<5)
 #define RBD_FEATURE_JOURNALING          (1ULL<<6)
 #define RBD_FEATURE_DATA_POOL           (1ULL<<7)
+#define RBD_FEATURE_OPERATIONS          (1ULL<<8)
 
 #define RBD_FEATURES_DEFAULT             (RBD_FEATURE_LAYERING | \
                                          RBD_FEATURE_EXCLUSIVE_LOCK | \
@@ -24,6 +25,7 @@
 #define RBD_FEATURE_NAME_DEEP_FLATTEN    "deep-flatten"
 #define RBD_FEATURE_NAME_JOURNALING      "journaling"
 #define RBD_FEATURE_NAME_DATA_POOL       "data-pool"
+#define RBD_FEATURE_NAME_OPERATIONS      "operations"
 
 /// features that make an image inaccessible for read or write by
 /// clients that don't understand them
@@ -37,7 +39,8 @@
 					 RBD_FEATURE_OBJECT_MAP     | \
                                          RBD_FEATURE_FAST_DIFF      | \
                                          RBD_FEATURE_DEEP_FLATTEN   | \
-                                         RBD_FEATURE_JOURNALING)
+                                         RBD_FEATURE_JOURNALING     | \
+                                         RBD_FEATURE_OPERATIONS)
 
 #define RBD_FEATURES_ALL          	(RBD_FEATURE_LAYERING       | \
 					 RBD_FEATURE_STRIPINGV2     | \
@@ -46,7 +49,8 @@
                                          RBD_FEATURE_FAST_DIFF      | \
                                          RBD_FEATURE_DEEP_FLATTEN   | \
                                          RBD_FEATURE_JOURNALING     | \
-                                         RBD_FEATURE_DATA_POOL)
+                                         RBD_FEATURE_DATA_POOL      | \
+                                         RBD_FEATURE_OPERATIONS)
 
 /// features that may be dynamically enabled or disabled
 #define RBD_FEATURES_MUTABLE            (RBD_FEATURE_EXCLUSIVE_LOCK | \
@@ -66,6 +70,17 @@
 
 /// features that will be implicitly enabled
 #define RBD_FEATURES_IMPLICIT_ENABLE  (RBD_FEATURE_STRIPINGV2 | \
-                                       RBD_FEATURE_DATA_POOL)
+                                       RBD_FEATURE_DATA_POOL | \
+                                       RBD_FEATURE_OPERATIONS)
+
+/// features that cannot be controlled by the user
+#define RBD_FEATURES_INTERNAL         (RBD_FEATURE_OPERATIONS)
+
+#define RBD_OPERATION_FEATURE_CLONE_V2        (1ULL<<0)
+
+#define RBD_OPERATION_FEATURE_NAME_CLONE_V2   "clone"
+
+/// all valid operation features
+#define RBD_OPERATION_FEATURES_ALL (RBD_OPERATION_FEATURE_CLONE_V2)
 
 #endif

src/pybind/mgr/dashboard/rbd_ls.py

@@ -64,6 +64,7 @@ def _format_bitmask(self, features):
             rbd.RBD_FEATURE_DEEP_FLATTEN: "deep-flatten",
             rbd.RBD_FEATURE_JOURNALING: "journaling",
             rbd.RBD_FEATURE_DATA_POOL: "data-pool",
+            rbd.RBD_FEATURE_OPERATIONS: "operations",
         }
 
         for key in RBD_FEATURES_NAME_MAPPING.keys():

src/pybind/rbd/rbd.pyx

@@ -57,6 +57,7 @@ cdef extern from "rbd/librbd.h" nogil:
         _RBD_FEATURE_DEEP_FLATTEN "RBD_FEATURE_DEEP_FLATTEN"
         _RBD_FEATURE_JOURNALING "RBD_FEATURE_JOURNALING"
         _RBD_FEATURE_DATA_POOL "RBD_FEATURE_DATA_POOL"
+        _RBD_FEATURE_OPERATIONS "RBD_FEATURE_OPERATIONS"
 
         _RBD_FEATURES_INCOMPATIBLE "RBD_FEATURES_INCOMPATIBLE"
         _RBD_FEATURES_RW_INCOMPATIBLE "RBD_FEATURES_RW_INCOMPATIBLE"
@@ -439,6 +440,7 @@ RBD_FEATURE_FAST_DIFF = _RBD_FEATURE_FAST_DIFF
 RBD_FEATURE_DEEP_FLATTEN = _RBD_FEATURE_DEEP_FLATTEN
 RBD_FEATURE_JOURNALING = _RBD_FEATURE_JOURNALING
 RBD_FEATURE_DATA_POOL = _RBD_FEATURE_DATA_POOL
+RBD_FEATURE_OPERATIONS = _RBD_FEATURE_OPERATIONS
 
 RBD_FEATURES_INCOMPATIBLE = _RBD_FEATURES_INCOMPATIBLE
 RBD_FEATURES_RW_INCOMPATIBLE = _RBD_FEATURES_RW_INCOMPATIBLE

src/test/cls_rbd/test_cls_rbd.cc

@@ -1511,6 +1511,7 @@ TEST_F(TestClsRbd, set_features)
                                   RBD_FEATURE_DEEP_FLATTEN));
 
   ASSERT_EQ(-EINVAL, set_features(&ioctx, oid, 0, RBD_FEATURE_LAYERING));
+  ASSERT_EQ(-EINVAL, set_features(&ioctx, oid, 0, RBD_FEATURE_OPERATIONS));
 }
 
 TEST_F(TestClsRbd, mirror) {
@@ -2499,3 +2500,49 @@ TEST_F(TestClsRbd, trash_methods)
 
   ioctx.close();
 }
+
+TEST_F(TestClsRbd, op_features)
+{
+  librados::IoCtx ioctx;
+  ASSERT_EQ(0, _rados.ioctx_create(_pool_name.c_str(), ioctx));
+
+  string oid = get_temp_image_name();
+  ASSERT_EQ(0, create_image(&ioctx, oid, 0, 22, 0, oid, -1));
+
+  uint64_t op_features = RBD_OPERATION_FEATURE_CLONE_V2;
+  uint64_t mask = ~RBD_OPERATION_FEATURES_ALL;
+  ASSERT_EQ(-EINVAL, op_features_set(&ioctx, oid, op_features, mask));
+
+  mask = 0;
+  ASSERT_EQ(0, op_features_set(&ioctx, oid, op_features, mask));
+
+  uint64_t actual_op_features;
+  ASSERT_EQ(0, op_features_get(&ioctx, oid, &actual_op_features));
+  ASSERT_EQ(0, actual_op_features);
+
+  uint64_t features;
+  ASSERT_EQ(0, get_features(&ioctx, oid, CEPH_NOSNAP, &features));
+  ASSERT_EQ(0u, features);
+
+  mask = RBD_OPERATION_FEATURES_ALL;
+  ASSERT_EQ(0, op_features_set(&ioctx, oid, op_features, mask));
+  ASSERT_EQ(0, op_features_get(&ioctx, oid, &actual_op_features));
+  ASSERT_EQ(mask, actual_op_features);
+
+  ASSERT_EQ(0, get_features(&ioctx, oid, CEPH_NOSNAP, &features));
+  ASSERT_EQ(RBD_FEATURE_OPERATIONS, features);
+
+  op_features = 0;
+  mask = RBD_OPERATION_FEATURE_CLONE_V2;
+  ASSERT_EQ(0, op_features_set(&ioctx, oid, op_features, mask));
+  ASSERT_EQ(0, op_features_get(&ioctx, oid, &actual_op_features));
+
+  uint64_t expected_op_features = RBD_OPERATION_FEATURES_ALL &
+                                    ~RBD_OPERATION_FEATURE_CLONE_V2;
+  ASSERT_EQ(expected_op_features, actual_op_features);
+
+  mask = 0;
+  ASSERT_EQ(0, op_features_set(&ioctx, oid, op_features, mask));
+  ASSERT_EQ(0, get_features(&ioctx, oid, CEPH_NOSNAP, &features));
+  ASSERT_EQ(0u, features);
+}

src/tools/rbd/ArgumentTypes.cc

@@ -26,6 +26,7 @@ const std::map<uint64_t, std::string> ImageFeatures::FEATURE_MAPPING = {
   {RBD_FEATURE_DEEP_FLATTEN, RBD_FEATURE_NAME_DEEP_FLATTEN},
   {RBD_FEATURE_JOURNALING, RBD_FEATURE_NAME_JOURNALING},
   {RBD_FEATURE_DATA_POOL, RBD_FEATURE_NAME_DATA_POOL},
+  {RBD_FEATURE_OPERATIONS, RBD_FEATURE_NAME_OPERATIONS},
 };
 
 Format::Formatter Format::create_formatter(bool pretty) const {