librbd/api/Mirror: return EINVAL from image_get_mode()

... when the image is disabled for mirroring.

When an image is disabled for mirroring, fetching the image's
mirroring mode is invalid. So, modify the Mirror::image_get_mode()
internal API to return EINVAL instead of success when mirroring is
disabled.

The Mirror::image_get_mode() method is called by the public C++, C, and
Python APIs that fetch the mirroring mode of an image. The behavior of
these public APIs will change. They will return an error code or raise
an exception indicating that it's an invalid operation to fetch the
image's mirroring mode when mirroring is disabled.

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

Author: ajarr

PendingReleaseNotes

@@ -155,6 +155,11 @@
   Related trackers:
    - https://tracker.ceph.com/issues/67777
 
+* RBD: Fetching the mirroring mode of an image is invalid if the image is
+  disabled for mirroring. The public APIs -- C++ `mirror_image_get_mode()`,
+  C `rbd_mirror_image_get_mode()`, and Python `Image.mirror_image_get_mode()`
+  -- will return EINVAL when mirroring is disabled.
+
 >=19.2.1
 
 * CephFS: Command `fs subvolume create` now allows tagging subvolumes through option

src/librbd/api/Mirror.cc

@@ -311,37 +311,29 @@ const char *pool_or_namespace(I *ictx) {
 
 struct C_ImageGetInfo : public Context {
   mirror_image_info_t *mirror_image_info;
-  mirror_image_mode_t *mirror_image_mode;
   Context *on_finish;
 
   cls::rbd::MirrorImage mirror_image;
   mirror::PromotionState promotion_state = mirror::PROMOTION_STATE_PRIMARY;
   std::string primary_mirror_uuid;
 
-  C_ImageGetInfo(mirror_image_info_t *mirror_image_info,
-                 mirror_image_mode_t *mirror_image_mode,  Context *on_finish)
-    : mirror_image_info(mirror_image_info),
-      mirror_image_mode(mirror_image_mode), on_finish(on_finish) {
+  C_ImageGetInfo(mirror_image_info_t *mirror_image_info, Context *on_finish)
+    : mirror_image_info(mirror_image_info), on_finish(on_finish) {
   }
 
   void finish(int r) override {
+    // Suppress ENOENT returned by GetInfoRequest when mirroring is
+    // disabled -- mirror_image.state will indicate that anyway.
     if (r < 0 && r != -ENOENT) {
       on_finish->complete(r);
       return;
     }
 
-    if (mirror_image_info != nullptr) {
-      mirror_image_info->global_id = mirror_image.global_image_id;
-      mirror_image_info->state = static_cast<rbd_mirror_image_state_t>(
-        mirror_image.state);
-      mirror_image_info->primary = (
-        promotion_state == mirror::PROMOTION_STATE_PRIMARY);
-    }
-
-    if (mirror_image_mode != nullptr) {
-      *mirror_image_mode =
-        static_cast<rbd_mirror_image_mode_t>(mirror_image.mode);
-    }
+    mirror_image_info->global_id = mirror_image.global_image_id;
+    mirror_image_info->state = static_cast<mirror_image_state_t>(
+      mirror_image.state);
+    mirror_image_info->primary = (
+      promotion_state == mirror::PROMOTION_STATE_PRIMARY);
 
     on_finish->complete(0);
   }
@@ -357,7 +349,7 @@ struct C_ImageGetGlobalStatus : public C_ImageGetInfo {
       const std::string &image_name,
       mirror_image_global_status_t *mirror_image_global_status,
       Context *on_finish)
-    : C_ImageGetInfo(&mirror_image_global_status->info, nullptr, on_finish),
+    : C_ImageGetInfo(&mirror_image_global_status->info, on_finish),
       image_name(image_name),
       mirror_image_global_status(mirror_image_global_status) {
   }
@@ -384,6 +376,36 @@ struct C_ImageGetGlobalStatus : public C_ImageGetInfo {
   }
 };
 
+struct C_ImageGetMode : public Context {
+  mirror_image_mode_t *mirror_image_mode;
+  Context *on_finish;
+
+  cls::rbd::MirrorImage mirror_image;
+  mirror::PromotionState promotion_state = mirror::PROMOTION_STATE_PRIMARY;
+  std::string primary_mirror_uuid;
+
+  C_ImageGetMode(mirror_image_mode_t *mirror_image_mode,  Context *on_finish)
+    : mirror_image_mode(mirror_image_mode), on_finish(on_finish) {
+  }
+
+  void finish(int r) override {
+    // Suppress ENOENT returned by GetInfoRequest when mirroring is
+    // disabled -- mirror_image.state will indicate that anyway.
+    if (r < 0 && r != -ENOENT) {
+      on_finish->complete(r);
+      return;
+    } else if (mirror_image.state == cls::rbd::MIRROR_IMAGE_STATE_DISABLED) {
+      on_finish->complete(-EINVAL);
+      return;
+    }
+
+    *mirror_image_mode =
+      static_cast<mirror_image_mode_t>(mirror_image.mode);
+
+    on_finish->complete(0);
+  }
+};
+
 template <typename I>
 struct C_ImageSnapshotCreate : public Context {
   I *ictx;
@@ -858,7 +880,7 @@ void Mirror<I>::image_get_info(I *ictx, mirror_image_info_t *mirror_image_info,
         return;
       }
 
-      auto ctx = new C_ImageGetInfo(mirror_image_info, nullptr, on_finish);
+      auto ctx = new C_ImageGetInfo(mirror_image_info, on_finish);
       auto req = mirror::GetInfoRequest<I>::create(*ictx, &ctx->mirror_image,
                                                    &ctx->promotion_state,
                                                    &ctx->primary_mirror_uuid,
@@ -895,7 +917,7 @@ void Mirror<I>::image_get_info(librados::IoCtx& io_ctx,
   ldout(cct, 20) << "pool_id=" << io_ctx.get_id() << ", image_id=" << image_id
                  << dendl;
 
-  auto ctx = new C_ImageGetInfo(mirror_image_info, nullptr, on_finish);
+  auto ctx = new C_ImageGetInfo(mirror_image_info, on_finish);
   auto req = mirror::GetInfoRequest<I>::create(io_ctx, op_work_queue, image_id,
                                                &ctx->mirror_image,
                                                &ctx->promotion_state,
@@ -924,7 +946,7 @@ void Mirror<I>::image_get_mode(I *ictx, mirror_image_mode_t *mode,
   CephContext *cct = ictx->cct;
   ldout(cct, 20) << "ictx=" << ictx << dendl;
 
-  auto ctx = new C_ImageGetInfo(nullptr, mode, on_finish);
+  auto ctx = new C_ImageGetMode(mode, on_finish);
   auto req = mirror::GetInfoRequest<I>::create(*ictx, &ctx->mirror_image,
                                                &ctx->promotion_state,
                                                &ctx->primary_mirror_uuid, ctx);

src/test/librbd/test_librbd.cc

@@ -11338,6 +11338,13 @@ TEST_F(TestLibRBD, CreateWithMirrorEnabled) {
   librbd::Image parent_image;
   ASSERT_EQ(0, rbd.open(ioctx, parent_image, parent_name.c_str(), NULL));
 
+  librbd::mirror_image_info_t mirror_image_info;
+  ASSERT_EQ(0, parent_image.mirror_image_get_info(&mirror_image_info,
+                                                  sizeof(mirror_image_info)));
+  ASSERT_EQ(RBD_MIRROR_IMAGE_ENABLED, mirror_image_info.state);
+  ASSERT_NE("", mirror_image_info.global_id);
+  ASSERT_EQ(true, mirror_image_info.primary);
+
   librbd::mirror_image_mode_t mirror_image_mode;
   ASSERT_EQ(0, parent_image.mirror_image_get_mode(&mirror_image_mode));
   ASSERT_EQ(RBD_MIRROR_IMAGE_MODE_SNAPSHOT, mirror_image_mode);
@@ -11358,6 +11365,14 @@ TEST_F(TestLibRBD, CreateWithMirrorEnabled) {
   ASSERT_EQ(0, child_image.mirror_image_disable(true));
   ASSERT_EQ(0, parent_image.mirror_image_disable(true));
   ASSERT_EQ(0, rbd.mirror_mode_set(ioctx, RBD_MIRROR_MODE_DISABLED));
+
+  ASSERT_EQ(0, parent_image.mirror_image_get_info(&mirror_image_info,
+                                                  sizeof(mirror_image_info)));
+  ASSERT_EQ(RBD_MIRROR_IMAGE_DISABLED, mirror_image_info.state);
+  ASSERT_EQ("", mirror_image_info.global_id);
+  ASSERT_EQ(false, mirror_image_info.primary);
+
+  ASSERT_EQ(-EINVAL, parent_image.mirror_image_get_mode(&mirror_image_mode));
 }
 
 TEST_F(TestLibRBD, FlushCacheWithCopyupOnExternalSnapshot) {

src/test/librbd/test_mirroring.cc

@@ -160,6 +160,10 @@ class TestMirroring : public TestFixture {
     ASSERT_EQ(0, image.mirror_image_get_info(&mirror_image, sizeof(mirror_image)));
     ASSERT_EQ(mirror_state, mirror_image.state);
 
+    librbd::mirror_image_mode_t mirror_image_mode;
+    ASSERT_EQ(mirror_state != RBD_MIRROR_IMAGE_DISABLED ? 0 : -EINVAL,
+              image.mirror_image_get_mode(&mirror_image_mode));
+
     librbd::mirror_image_global_status_t status;
     ASSERT_EQ(0, image.mirror_image_get_global_status(&status, sizeof(status)));
     librbd::mirror_image_site_status_t local_status;

src/test/pybind/test_rbd.py

@@ -2678,6 +2678,7 @@ def test_mirror_image(self):
         self.image.mirror_image_disable(True)
         info = self.image.mirror_image_get_info()
         self.check_info(info, '', RBD_MIRROR_IMAGE_DISABLED, False)
+        assert_raises(InvalidArgument, self.image.mirror_image_get_mode)
 
         self.image.mirror_image_enable()
         info = self.image.mirror_image_get_info()
@@ -2835,15 +2836,37 @@ def test_aio_mirror_image_create_snapshot(self):
         peer_uuid = self.rbd.mirror_peer_add(ioctx, "cluster", "client")
         self.rbd.mirror_mode_set(ioctx, RBD_MIRROR_MODE_IMAGE)
         self.image.mirror_image_disable(False)
-        self.image.mirror_image_enable(RBD_MIRROR_IMAGE_MODE_SNAPSHOT)
 
+        # this is a list so that the local cb() can modify it
+        info = [123]
+        def cb(_, _info):
+            info[0] = _info
+
+        comp = self.image.aio_mirror_image_get_info(cb)
+        comp.wait_for_complete_and_cb()
+        assert_not_equal(info[0], None)
+        eq(comp.get_return_value(), 0)
+        eq(sys.getrefcount(comp), 2)
+        info = info[0]
+        self.check_info(info, "", RBD_MIRROR_IMAGE_DISABLED, False)
+
+        mode = [123]
+        def cb(_, _mode):
+            mode[0] = _mode
+
+        comp = self.image.aio_mirror_image_get_mode(cb)
+        comp.wait_for_complete_and_cb()
+        eq(comp.get_return_value(), -errno.EINVAL)
+        eq(sys.getrefcount(comp), 2)
+        eq(mode[0], None)
+
+        self.image.mirror_image_enable(RBD_MIRROR_IMAGE_MODE_SNAPSHOT)
         snaps = list(self.image.list_snaps())
         eq(1, len(snaps))
         snap = snaps[0]
         eq(snap['namespace'], RBD_SNAP_NAMESPACE_TYPE_MIRROR)
         eq(RBD_SNAP_MIRROR_STATE_PRIMARY, snap['mirror']['state'])
 
-        # this is a list so that the local cb() can modify it
         info = [None]
         def cb(_, _info):
             info[0] = _info