jewel: rbd-mirror: do not re-use image id from mirror directory if creating image

src/cls/rbd/cls_rbd.cc

@@ -3112,6 +3112,12 @@ int image_set(cls_method_context_t hctx, const string &image_id,
               cpp_strerror(r).c_str());
       return r;
     }
+
+    // make sure this was not a race for disabling
+    if (mirror_image.state == cls::rbd::MIRROR_IMAGE_STATE_DISABLING) {
+      CLS_ERR("image '%s' is already disabled", image_id.c_str());
+      return r;
+    }
   } else if (r < 0) {
     CLS_ERR("error reading mirrored image '%s': '%s'", image_id.c_str(),
 	    cpp_strerror(r).c_str());

src/test/cls_rbd/test_cls_rbd.cc

@@ -1383,6 +1383,10 @@ TEST_F(TestClsRbd, mirror_image) {
   cls::rbd::MirrorImage image3("uuid3", cls::rbd::MIRROR_IMAGE_STATE_ENABLED);
 
   ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id1", image1));
+  ASSERT_EQ(-ENOENT, mirror_image_set(&ioctx, "image_id2", image2));
+  image2.state = cls::rbd::MIRROR_IMAGE_STATE_ENABLED;
+  ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id2", image2));
+  image2.state = cls::rbd::MIRROR_IMAGE_STATE_DISABLING;
   ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id2", image2));
   ASSERT_EQ(-EINVAL, mirror_image_set(&ioctx, "image_id1", image2));
   ASSERT_EQ(-EEXIST, mirror_image_set(&ioctx, "image_id3", image2));

src/test/rbd_mirror/test_ImageDeleter.cc

@@ -342,6 +342,11 @@ TEST_F(TestImageDeleter, Delete_Image_With_Clone) {
 TEST_F(TestImageDeleter, Delete_NonExistent_Image) {
   remove_image();
 
+  cls::rbd::MirrorImage mirror_image(GLOBAL_IMAGE_ID,
+                              MirrorImageState::MIRROR_IMAGE_STATE_ENABLED);
+  EXPECT_EQ(0, cls_client::mirror_image_set(&m_local_io_ctx, m_local_image_id,
+                                            mirror_image));
+
   m_deleter->schedule_image_delete(m_local_pool_id, m_local_image_id,
       m_image_name, GLOBAL_IMAGE_ID);
 
@@ -358,6 +363,14 @@ TEST_F(TestImageDeleter, Delete_NonExistent_Image) {
 TEST_F(TestImageDeleter, Delete_NonExistent_Image_With_MirroringState) {
   remove_image(true);
 
+  cls::rbd::MirrorImage mirror_image(GLOBAL_IMAGE_ID,
+                              MirrorImageState::MIRROR_IMAGE_STATE_ENABLED);
+  EXPECT_EQ(0, cls_client::mirror_image_set(&m_local_io_ctx, m_local_image_id,
+                                            mirror_image));
+  mirror_image.state = MirrorImageState::MIRROR_IMAGE_STATE_DISABLING;
+  EXPECT_EQ(0, cls_client::mirror_image_set(&m_local_io_ctx, m_local_image_id,
+                                            mirror_image));
+
   m_deleter->schedule_image_delete(m_local_pool_id, m_local_image_id,
       m_image_name, GLOBAL_IMAGE_ID);
 
@@ -371,6 +384,22 @@ TEST_F(TestImageDeleter, Delete_NonExistent_Image_With_MirroringState) {
   check_image_deleted();
 }
 
+TEST_F(TestImageDeleter, Delete_NonExistent_Image_Without_MirroringState) {
+  remove_image();
+
+  m_deleter->schedule_image_delete(m_local_pool_id, m_local_image_id,
+      m_image_name, GLOBAL_IMAGE_ID);
+
+  C_SaferCond ctx;
+  m_deleter->wait_for_scheduled_deletion(m_image_name, &ctx);
+  EXPECT_EQ(-ENOENT, ctx.wait());
+
+  ASSERT_EQ(0u, m_deleter->get_delete_queue_items().size());
+  ASSERT_EQ(0u, m_deleter->get_failed_queue_items().size());
+
+  check_image_deleted();
+}
+
 TEST_F(TestImageDeleter, Fail_Delete_NonPrimary_Image) {
   ImageCtx *ictx = new ImageCtx("", m_local_image_id, "", m_local_io_ctx,
                                 false);

src/tools/rbd_mirror/ImageDeleter.cc

@@ -281,7 +281,14 @@ bool ImageDeleter::process_image_delete() {
   mirror_image.state = cls::rbd::MIRROR_IMAGE_STATE_DISABLING;
   r = cls_client::mirror_image_set(&ioctx, curr_deletion->local_image_id,
                                            mirror_image);
-  if (r == -EEXIST || r == -EINVAL) {
+  if (r == -ENOENT) {
+    dout(10) << "local image is not mirrored, aborting deletion..." << dendl;
+    m_delete_lock.Lock();
+    DeleteInfo *del_info = curr_deletion.release();
+    m_delete_lock.Unlock();
+    del_info->notify(r);
+    return true;
+  } else if (r == -EEXIST || r == -EINVAL) {
     derr << "cannot disable mirroring for image id" << curr_deletion->local_image_id
          << ": global_image_id has changed/reused, aborting deletion: "
          << cpp_strerror(r) << dendl;

src/tools/rbd_mirror/image_replayer/BootstrapRequest.cc

@@ -357,6 +357,7 @@ template <typename I>
 void BootstrapRequest<I>::create_local_image() {
   dout(20) << dendl;
 
+  m_local_image_id = "";
   update_progress("CREATE_LOCAL_IMAGE");
 
   Context *ctx = create_context_callback<