librbd/image: create rbd_trash object during RBD pool initialization

... and RBD namespace creation.

It was not possible to remove a RBD image when OSDs were full and the
'rbd_trash' object was not already created in the image's pool or pool
namespace. The 'rbd_trash' object was created in a pool or namespace
during the first instance of image removal from the pool or namespace.
If no images were ever removed from a RBD pool or namespace and the
OSDs became full, removal of images using the CLI failed. The failure
occured when trying to move the images to trash since the 'rbd_trash'
object was missing in the pool or namespace.

Fix this issue by creating the rbd_trash object in a pool when
initalizing the pool as a RBD pool and when creating a RBD namespace.

Fixes: https://tracker.ceph.com/issues/64800
Signed-off-by: Ramana Raja <[email protected]>

Author: ajarr

src/librbd/api/Namespace.cc

@@ -64,29 +64,47 @@ int Namespace<I>::create(librados::IoCtx& io_ctx, const std::string& name)
     return r;
   }
 
+  int ret_val;
   librados::IoCtx ns_ctx;
   ns_ctx.dup(io_ctx);
   ns_ctx.set_namespace(name);
 
+  r = ns_ctx.create(RBD_TRASH, false);
+  if (r < 0) {
+    lderr(cct) << "failed to create trash: " << cpp_strerror(r) << dendl;
+    goto remove_namespace;
+  }
+
   r = cls_client::dir_state_set(&ns_ctx, RBD_DIRECTORY,
                                 cls::rbd::DIRECTORY_STATE_READY);
   if (r < 0) {
     lderr(cct) << "failed to initialize image directory: " << cpp_strerror(r)
                << dendl;
-    goto rollback;
+    goto remove_dir_and_trash;
   }
 
   return 0;
 
-rollback:
-  int ret_val = ns_ctx.remove(RBD_DIRECTORY);
+remove_dir_and_trash:
+  // dir_state_set method may fail before or after implicitly creating
+  // rbd_directory object
+  ret_val = ns_ctx.remove(RBD_DIRECTORY);
   if (ret_val < 0 && ret_val != -ENOENT) {
-    lderr(cct) << "failed to remove image directory: " << cpp_strerror(ret_val) << dendl;
+    lderr(cct) << "failed to remove image directory: " << cpp_strerror(ret_val)
+               << dendl;
   }
 
+  ret_val = ns_ctx.remove(RBD_TRASH);
+  if (ret_val < 0) {
+    lderr(cct) << "failed to remove trash: " << cpp_strerror(ret_val)
+               << dendl;
+  }
+
+remove_namespace:
   ret_val = cls_client::namespace_remove(&default_ns_ctx, name);
   if (ret_val < 0) {
-    lderr(cct) << "failed to remove namespace: " << cpp_strerror(ret_val) << dendl;
+    lderr(cct) << "failed to remove namespace: " << cpp_strerror(ret_val)
+               << dendl;
   }
 
   return r;

src/librbd/api/Pool.cc

@@ -243,6 +243,14 @@ int Pool<I>::init(librados::IoCtx& io_ctx, bool force) {
 
   int r = io_ctx.application_enable(pg_pool_t::APPLICATION_NAME_RBD, force);
   if (r < 0) {
+    lderr(cct) << "failed to enable RBD application: " << cpp_strerror(r)
+               << dendl;
+    return r;
+  }
+
+  r = io_ctx.create(RBD_TRASH, false);
+  if (r < 0) {
+    lderr(cct) << "failed to create trash: " << cpp_strerror(r) << dendl;
     return r;
   }
 

src/test/librbd/test_librbd.cc

@@ -2172,7 +2172,32 @@ static void remove_full_try(rados_ioctx_t ioctx, const std::string& image_name,
 
   ASSERT_EQ(0, rbd_close(image));
 
-  // make sure we have latest map that marked the pool full
+  // If using a separate data pool, also make the metadata pool full
+  char pool_name[80];
+  ASSERT_GT(rados_ioctx_get_pool_name(ioctx, pool_name, sizeof(pool_name)), 0);
+  if (pool_name != data_pool_name) {
+    struct rados_pool_stat_t st;
+    ASSERT_EQ(0, rados_ioctx_pool_stat(ioctx, &st));
+    cmdstr = "{\"prefix\": \"osd pool set-quota\", \"pool\": \"" +
+        std::string(pool_name) + "\", \"field\": \"max_objects\", \"val\": \"" +
+        std::to_string(st.num_objects) + "\"}";
+    cmd[0] = (char *)cmdstr.c_str();
+    ASSERT_EQ(0, rados_mon_command(rados_ioctx_get_cluster(ioctx),
+                                   (const char **)cmd, 1, "", 0, nullptr, 0,
+                                   nullptr, 0));
+
+    for (int i = 0; i < 50; i++) {
+      auto temp_image_name = TestLibRBD::get_temp_image_name();
+      ret = create_image(ioctx, temp_image_name.c_str(), size, &order);
+      if (ret < 0) {
+        break;
+      }
+      sleep(1);
+    }
+    ASSERT_EQ(ret, -EDQUOT);
+  }
+
+  // make sure we have latest map that marked the pool(s) full
   ASSERT_EQ(0, rados_wait_for_latest_osdmap(rados_ioctx_get_cluster(ioctx)));
   ASSERT_EQ(0, rbd_remove(ioctx, image_name.c_str()));
 }
@@ -2185,17 +2210,13 @@ TEST_F(TestLibRBD, RemoveFullTry)
   rados_ioctx_t ioctx;
   auto pool_name = create_pool(true);
   ASSERT_EQ(0, rados_ioctx_create(_cluster, pool_name.c_str(), &ioctx));
+  ASSERT_EQ(0, rbd_pool_init(ioctx, true));
   // cancel out rbd_default_data_pool -- we need an image without
   // a separate data pool
   ASSERT_EQ(0, rbd_pool_metadata_set(ioctx, "conf_rbd_default_data_pool",
                                      pool_name.c_str()));
 
-  int order = 0;
   auto image_name = get_temp_image_name();
-  // FIXME: this is a workaround for rbd_trash object being created
-  // on the first remove -- pre-create it to avoid bumping into quota
-  ASSERT_EQ(0, create_image(ioctx, image_name.c_str(), 0, &order));
-  ASSERT_EQ(0, rbd_remove(ioctx, image_name.c_str()));
   remove_full_try(ioctx, image_name, pool_name);
 
   rados_ioctx_destroy(ioctx);
@@ -2207,12 +2228,56 @@ TEST_F(TestLibRBD, RemoveFullTryDataPool)
   REQUIRE(!is_rbd_pwl_enabled((CephContext *)_rados.cct()));
   REQUIRE(!is_librados_test_stub(_rados));
 
+  rados_ioctx_t ioctx;
+  auto pool_name = create_pool(true);
+  auto data_pool_name = create_pool(true);
+  ASSERT_EQ(0, rados_ioctx_create(_cluster, pool_name.c_str(), &ioctx));
+  ASSERT_EQ(0, rbd_pool_init(ioctx, true));
+  ASSERT_EQ(0, rbd_pool_metadata_set(ioctx, "conf_rbd_default_data_pool",
+                                     data_pool_name.c_str()));
+
+  auto image_name = get_temp_image_name();
+  remove_full_try(ioctx, image_name, data_pool_name);
+
+  rados_ioctx_destroy(ioctx);
+}
+
+TEST_F(TestLibRBD, RemoveFullTryNamespace)
+{
+  REQUIRE_FORMAT_V2();
+  REQUIRE(!is_rbd_pwl_enabled((CephContext *)_rados.cct()));
+  REQUIRE(!is_librados_test_stub(_rados));
+
+  rados_ioctx_t ioctx;
+  auto pool_name = create_pool(true);
+  ASSERT_EQ(0, rados_ioctx_create(_cluster, pool_name.c_str(), &ioctx));
+  // cancel out rbd_default_data_pool -- we need an image without
+  // a separate data pool
+  ASSERT_EQ(0, rbd_pool_metadata_set(ioctx, "conf_rbd_default_data_pool",
+                                     pool_name.c_str()));
+  ASSERT_EQ(0, rbd_namespace_create(ioctx, "name1"));
+  rados_ioctx_set_namespace(ioctx, "name1");
+
+  auto image_name = get_temp_image_name();
+  remove_full_try(ioctx, image_name, pool_name);
+
+  rados_ioctx_destroy(ioctx);
+}
+
+TEST_F(TestLibRBD, RemoveFullTryNamespaceDataPool)
+{
+  REQUIRE_FORMAT_V2();
+  REQUIRE(!is_rbd_pwl_enabled((CephContext *)_rados.cct()));
+  REQUIRE(!is_librados_test_stub(_rados));
+
   rados_ioctx_t ioctx;
   auto pool_name = create_pool(true);
   auto data_pool_name = create_pool(true);
   ASSERT_EQ(0, rados_ioctx_create(_cluster, pool_name.c_str(), &ioctx));
   ASSERT_EQ(0, rbd_pool_metadata_set(ioctx, "conf_rbd_default_data_pool",
                                      data_pool_name.c_str()));
+  ASSERT_EQ(0, rbd_namespace_create(ioctx, "name1"));
+  rados_ioctx_set_namespace(ioctx, "name1");
 
   auto image_name = get_temp_image_name();
   remove_full_try(ioctx, image_name, data_pool_name);