Merge pull request ceph#58180 from guojidan/format

rbd: add the validate of the format and clone_format

Reviewed-by: Ilya Dryomov <[email protected]>

Author: idryomov

src/librbd/api/Image.cc

@@ -570,8 +570,8 @@ int Image<I>::deep_copy(I *src, librados::IoCtx& dest_md_ctx,
   if (opts.get(RBD_IMAGE_OPTION_FORMAT, &format) != 0) {
     opts.set(RBD_IMAGE_OPTION_FORMAT, format);
   }
-  if (format == 1) {
-    lderr(cct) << "old format not supported for destination image" << dendl;
+  if (format != 2) {
+    lderr(cct) << "unsupported destination image format: " << format << dendl;
     return -EINVAL;
   }
   uint64_t stripe_unit = src->stripe_unit;

src/librbd/image/CloneRequest.cc

@@ -87,8 +87,8 @@ void CloneRequest<I>::validate_options() {
 
   uint64_t format = 0;
   m_opts.get(RBD_IMAGE_OPTION_FORMAT, &format);
-  if (format < 2) {
-    lderr(m_cct) << "format 2 or later required for clone" << dendl;
+  if (format != 2) {
+    lderr(m_cct) << "unsupported format for clone: " << format << dendl;
     complete(-EINVAL);
     return;
   }
@@ -124,6 +124,12 @@ void CloneRequest<I>::validate_options() {
     }
   }
 
+  if (m_clone_format < 1 || m_clone_format > 2) {
+    lderr(m_cct) << "unsupported clone format: " << m_clone_format << dendl;
+    complete(-EINVAL);
+    return;
+  }
+
   if (m_clone_format == 1 &&
       m_parent_io_ctx.get_namespace() != m_ioctx.get_namespace()) {
     ldout(m_cct, 1) << "clone v2 required for cross-namespace clones" << dendl;

src/librbd/internal.cc

@@ -638,6 +638,11 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
     uint64_t format;
     if (opts.get(RBD_IMAGE_OPTION_FORMAT, &format) != 0)
       format = cct->_conf.get_val<uint64_t>("rbd_default_format");
+
+    if (format < 1 || format > 2) {
+      lderr(cct) << "unsupported format: " << format << dendl;
+      return -EINVAL;
+    }
     bool old_format = format == 1;
 
     // make sure it doesn't already exist, in either format

src/test/librbd/test_librbd.cc

@@ -13625,6 +13625,171 @@ TEST_F(TestLibRBD, ConcurrentOperations)
   ioctx.close();
 }
 
+TEST_F(TestLibRBD, FormatAndCloneFormatOptions)
+{
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librados::IoCtx ioctx;
+  ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
+
+  librbd::ImageOptions opts_with_0;
+  ASSERT_EQ(0, opts_with_0.set(RBD_IMAGE_OPTION_FORMAT, 0));
+  librbd::ImageOptions opts_with_1;
+  ASSERT_EQ(0, opts_with_1.set(RBD_IMAGE_OPTION_FORMAT, 1));
+  librbd::ImageOptions opts_with_2;
+  ASSERT_EQ(0, opts_with_2.set(RBD_IMAGE_OPTION_FORMAT, 2));
+  librbd::ImageOptions opts_with_3;
+  ASSERT_EQ(0, opts_with_3.set(RBD_IMAGE_OPTION_FORMAT, 3));
+
+  uint64_t features;
+  ASSERT_TRUE(get_features(&features));
+  ASSERT_EQ(0, opts_with_2.set(RBD_IMAGE_OPTION_FEATURES, features));
+
+  // create
+  librbd::RBD rbd;
+  std::string name1 = get_temp_image_name();
+  std::string name2 = get_temp_image_name();
+  auto do_create = [&rbd, &ioctx](const auto& name, const auto& opts) {
+    auto mod_opts = opts;
+    return rbd.create4(ioctx, name.c_str(), 2 << 20, mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_create(name1, opts_with_0));
+  ASSERT_EQ(-EINVAL, do_create(name1, opts_with_3));
+  ASSERT_EQ(0, do_create(name1, opts_with_1));
+  auto verify_format_1 = [&rbd, &ioctx](const auto& name) {
+    librbd::Image image;
+    ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
+    uint8_t old_format;
+    ASSERT_EQ(0, image.old_format(&old_format));
+    ASSERT_TRUE(old_format);
+  };
+  ASSERT_NO_FATAL_FAILURE(verify_format_1(name1));
+  ASSERT_EQ(0, do_create(name2, opts_with_2));
+  auto verify_format_2 = [&rbd, &ioctx](const auto& name) {
+    librbd::Image image;
+    ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
+    uint8_t old_format;
+    ASSERT_EQ(0, image.old_format(&old_format));
+    ASSERT_FALSE(old_format);
+  };
+  ASSERT_NO_FATAL_FAILURE(verify_format_2(name2));
+
+  {
+    librbd::Image image;
+    ASSERT_EQ(0, rbd.open(ioctx, image, name2.c_str(), NULL));
+    ASSERT_EQ(0, image.snap_create("parent_snap"));
+    ASSERT_EQ(0, image.snap_protect("parent_snap"));
+  }
+
+  // clone
+  std::string clone_name1 = get_temp_image_name();
+  std::string clone_name2 = get_temp_image_name();
+  auto do_clone = [&rbd, &ioctx, &name2](const auto& clone_name,
+                                         const auto& opts) {
+    auto mod_opts = opts;
+    return rbd.clone3(ioctx, name2.c_str(), "parent_snap", ioctx,
+                      clone_name.c_str(), mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_clone(clone_name1, opts_with_0));
+  ASSERT_EQ(-EINVAL, do_clone(clone_name1, opts_with_1));
+  ASSERT_EQ(-EINVAL, do_clone(clone_name1, opts_with_3));
+  // if RBD_IMAGE_OPTION_CLONE_FORMAT isn't set, rbd_default_clone_format
+  // config option kicks in -- we aren't interested in its behavior here
+  ASSERT_EQ(0, do_clone(clone_name1, opts_with_2));
+  ASSERT_EQ(0, rbd.remove(ioctx, clone_name1.c_str()));
+
+  auto clone_opts_with_0 = opts_with_2;
+  ASSERT_EQ(0, clone_opts_with_0.set(RBD_IMAGE_OPTION_CLONE_FORMAT, 0));
+  auto clone_opts_with_1 = opts_with_2;
+  ASSERT_EQ(0, clone_opts_with_1.set(RBD_IMAGE_OPTION_CLONE_FORMAT, 1));
+  auto clone_opts_with_2 = opts_with_2;
+  ASSERT_EQ(0, clone_opts_with_2.set(RBD_IMAGE_OPTION_CLONE_FORMAT, 2));
+  auto clone_opts_with_3 = opts_with_2;
+  ASSERT_EQ(0, clone_opts_with_3.set(RBD_IMAGE_OPTION_CLONE_FORMAT, 3));
+
+  ASSERT_EQ(-EINVAL, do_clone(clone_name1, clone_opts_with_0));
+  ASSERT_EQ(-EINVAL, do_clone(clone_name1, clone_opts_with_3));
+  ASSERT_EQ(0, do_clone(clone_name1, clone_opts_with_1));
+  {
+    librbd::Image image;
+    ASSERT_EQ(0, rbd.open(ioctx, image, clone_name1.c_str(), NULL));
+    uint64_t op_features;
+    ASSERT_EQ(0, image.get_op_features(&op_features));
+    ASSERT_EQ(op_features, 0);
+  }
+  ASSERT_EQ(0, do_clone(clone_name2, clone_opts_with_2));
+  {
+    librbd::Image image;
+    ASSERT_EQ(0, rbd.open(ioctx, image, clone_name2.c_str(), NULL));
+    uint64_t op_features;
+    ASSERT_EQ(0, image.get_op_features(&op_features));
+    ASSERT_EQ(op_features, RBD_OPERATION_FEATURE_CLONE_CHILD);
+  }
+
+  librbd::Image image;
+  ASSERT_EQ(0, rbd.open(ioctx, image, name1.c_str(), NULL));
+
+  // copy
+  std::string copy_name1 = get_temp_image_name();
+  std::string copy_name2 = get_temp_image_name();
+  auto do_copy = [&image, &ioctx](const auto& copy_name, const auto& opts) {
+    auto mod_opts = opts;
+    return image.copy3(ioctx, copy_name.c_str(), mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_copy(copy_name1, opts_with_0));
+  ASSERT_EQ(-EINVAL, do_copy(copy_name1, opts_with_3));
+  ASSERT_EQ(0, do_copy(copy_name1, opts_with_1));
+  ASSERT_NO_FATAL_FAILURE(verify_format_1(copy_name1));
+  ASSERT_EQ(0, do_copy(copy_name2, opts_with_2));
+  ASSERT_NO_FATAL_FAILURE(verify_format_2(copy_name2));
+
+  // deep copy
+  std::string deep_copy_name = get_temp_image_name();
+  auto do_deep_copy = [&image, &ioctx, &deep_copy_name](const auto& opts) {
+    auto mod_opts = opts;
+    return image.deep_copy(ioctx, deep_copy_name.c_str(), mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_deep_copy(opts_with_0));
+  ASSERT_EQ(-EINVAL, do_deep_copy(opts_with_1));
+  ASSERT_EQ(-EINVAL, do_deep_copy(opts_with_3));
+  ASSERT_EQ(0, do_deep_copy(opts_with_2));
+  ASSERT_NO_FATAL_FAILURE(verify_format_2(deep_copy_name));
+
+  ASSERT_EQ(0, image.close());
+
+  // migration
+  std::string migrate_name = get_temp_image_name();
+  auto do_migrate = [&rbd, &ioctx, &name1, &migrate_name](const auto& opts) {
+    auto mod_opts = opts;
+    return rbd.migration_prepare(ioctx, name1.c_str(), ioctx,
+                                 migrate_name.c_str(), mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_migrate(opts_with_0));
+  ASSERT_EQ(-EINVAL, do_migrate(opts_with_1));
+  ASSERT_EQ(-EINVAL, do_migrate(opts_with_3));
+  ASSERT_EQ(0, do_migrate(opts_with_2));
+  ASSERT_NO_FATAL_FAILURE(verify_format_2(migrate_name));
+
+  // import-only migration
+  std::string source_spec = R"({
+    "type": "native",
+    "pool_name": ")" + m_pool_name + R"(",
+    "image_name": ")" + name2 + R"(",
+    "snap_name": "parent_snap"
+})";
+  std::string import_name = get_temp_image_name();
+  auto do_migrate_import = [&rbd, &ioctx, &source_spec, &import_name](
+      const auto& opts) {
+    auto mod_opts = opts;
+    return rbd.migration_prepare_import(source_spec.c_str(), ioctx,
+                                        import_name.c_str(), mod_opts);
+  };
+  ASSERT_EQ(-EINVAL, do_migrate_import(opts_with_0));
+  ASSERT_EQ(-EINVAL, do_migrate_import(opts_with_1));
+  ASSERT_EQ(-EINVAL, do_migrate_import(opts_with_3));
+  ASSERT_EQ(0, do_migrate_import(opts_with_2));
+  ASSERT_NO_FATAL_FAILURE(verify_format_2(import_name));
+}
 
 // poorman's ceph_assert()
 namespace ceph {

src/test/pybind/test_rbd.py

@@ -1730,6 +1730,11 @@ def test_stripe_unit_and_count(self):
 
     def test_clone_format(self):
         clone_name2 = get_temp_image_name()
+        assert_raises(InvalidArgument, self.rbd.clone, ioctx, image_name,
+                       'snap1', ioctx, clone_name2, features, clone_format=0)
+        assert_raises(InvalidArgument, self.rbd.clone, ioctx, image_name,
+                       'snap1', ioctx, clone_name2, features, clone_format=3)
+
         self.rbd.clone(ioctx, image_name, 'snap1', ioctx, clone_name2,
                        features, clone_format=1)
         with Image(ioctx, clone_name2) as clone2: