librbd: permit removal of image being bootstrapped by rbd-mirror

qa/workunits/rbd/rbd_mirror.sh

@@ -223,8 +223,6 @@ unprotect_snapshot ${CLUSTER2} ${POOL} ${image5} 'snap2'
 for i in ${image3} ${image5}; do
   remove_snapshot ${CLUSTER2} ${POOL} ${i} 'snap1'
   remove_snapshot ${CLUSTER2} ${POOL} ${i} 'snap2'
-  # workaround #16555: before removing make sure it is not still bootstrapped
-  wait_for_image_replay_started ${CLUSTER1} ${POOL} ${i}
   remove_image_retry ${CLUSTER2} ${POOL} ${i}
 done
 

src/librbd/internal.cc

@@ -163,6 +163,46 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force,
   return 0;
 }
 
+void filter_out_mirror_watchers(ImageCtx *ictx,
+                                std::list<obj_watch_t> *watchers) {
+  if (watchers->empty()) {
+    return;
+  }
+
+  if ((ictx->features & RBD_FEATURE_JOURNALING) == 0) {
+    return;
+  }
+
+  cls::rbd::MirrorImage mirror_image;
+  int r = cls_client::mirror_image_get(&ictx->md_ctx, ictx->id, &mirror_image);
+  if (r < 0) {
+    if (r != -ENOENT) {
+      lderr(ictx->cct) << "failed to retrieve mirroring state: "
+                       << cpp_strerror(r) << dendl;
+    }
+    return;
+  }
+
+  if (mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+    return;
+  }
+
+  std::list<obj_watch_t> mirror_watchers;
+  r = ictx->md_ctx.list_watchers(RBD_MIRRORING, &mirror_watchers);
+  if (r < 0) {
+    if (r != -ENOENT) {
+      lderr(ictx->cct) << "error listing mirroring watchers: "
+                       << cpp_strerror(r) << dendl;
+    }
+    return;
+  }
+  for (auto &watcher : mirror_watchers) {
+    watchers->remove_if([watcher] (obj_watch_t &w) {
+        return (strncmp(w.addr, watcher.addr, sizeof(w.addr)) == 0);
+      });
+  }
+}
+
 } // anonymous namespace
 
   int detect_format(IoCtx &io_ctx, const string &name,
@@ -1554,6 +1594,12 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force,
         ictx->state->close();
         return r;
       }
+
+      // If an image is being bootstrapped by rbd-mirror, it implies
+      // that the rbd-mirror daemon currently has the image open.
+      // Permit removal if this is the case.
+      filter_out_mirror_watchers(ictx, &watchers);
+
       if (watchers.size() > 1) {
         lderr(cct) << "image has watchers - not removing" << dendl;
 	ictx->owner_lock.put_read();

src/test/librbd/test_mirroring.cc

@@ -644,3 +644,40 @@ TEST_F(TestMirroring, MirrorStatusList) {
   ASSERT_EQ(0, m_rbd.mirror_image_status_list(m_ioctx, last_read, 4096, &images));
   ASSERT_EQ(0U, images.size());
 }
+
+TEST_F(TestMirroring, RemoveBootstrapped)
+{
+  ASSERT_EQ(0, m_rbd.mirror_mode_set(m_ioctx, RBD_MIRROR_MODE_POOL));
+
+  uint64_t features = RBD_FEATURE_EXCLUSIVE_LOCK | RBD_FEATURE_JOURNALING;
+  int order = 20;
+  ASSERT_EQ(0, m_rbd.create2(m_ioctx, image_name.c_str(), 4096, features,
+                             &order));
+  librbd::Image image;
+  ASSERT_EQ(0, m_rbd.open(m_ioctx, image, image_name.c_str()));
+  ASSERT_EQ(-EBUSY, m_rbd.remove(m_ioctx, image_name.c_str()));
+
+  // simulate the image is open by rbd-mirror bootstrap
+  uint64_t handle;
+  struct MirrorWatcher : public librados::WatchCtx2 {
+    MirrorWatcher(librados::IoCtx &ioctx) : m_ioctx(ioctx) {
+    }
+    virtual void handle_notify(uint64_t notify_id, uint64_t cookie,
+                               uint64_t notifier_id, bufferlist& bl) {
+      // received IMAGE_UPDATED notification from remove
+      m_notified = true;
+      m_ioctx.notify_ack(RBD_MIRRORING, notify_id, cookie, bl);
+    }
+    virtual void handle_error(uint64_t cookie, int err) {
+    }
+    librados::IoCtx &m_ioctx;
+    bool m_notified = false;
+  } watcher(m_ioctx);
+  ASSERT_EQ(0, m_ioctx.create(RBD_MIRRORING, false));
+  ASSERT_EQ(0, m_ioctx.watch2(RBD_MIRRORING, &handle, &watcher));
+  // now remove should succeed
+  ASSERT_EQ(0, m_rbd.remove(m_ioctx, image_name.c_str()));
+  ASSERT_EQ(0, m_ioctx.unwatch2(handle));
+  ASSERT_TRUE(watcher.m_notified);
+  ASSERT_EQ(0, image.close());
+}