Merge pull request ceph#61129 from idryomov/wip-68998

librbd: avoid data corruption on flatten when object map is inconsistent

Reviewed-by: Ramana Raja <[email protected]>

src/librbd/ObjectMap.cc

@@ -106,32 +106,6 @@ bool ObjectMap<I>::object_may_exist(uint64_t object_no) const
   return exists;
 }
 
-template <typename I>
-bool ObjectMap<I>::object_may_not_exist(uint64_t object_no) const
-{
-  ceph_assert(ceph_mutex_is_locked(m_image_ctx.image_lock));
-
-  // Fall back to default logic if object map is disabled or invalid
-  if (!m_image_ctx.test_features(RBD_FEATURE_OBJECT_MAP,
-                                 m_image_ctx.image_lock)) {
-    return true;
-  }
-
-  bool flags_set;
-  int r = m_image_ctx.test_flags(m_image_ctx.snap_id,
-                                 RBD_FLAG_OBJECT_MAP_INVALID,
-                                 m_image_ctx.image_lock, &flags_set);
-  if (r < 0 || flags_set) {
-    return true;
-  }
-
-  uint8_t state = (*this)[object_no];
-  bool nonexistent = (state != OBJECT_EXISTS && state != OBJECT_EXISTS_CLEAN);
-  ldout(m_image_ctx.cct, 20) << "object_no=" << object_no << " r="
-                             << nonexistent << dendl;
-  return nonexistent;
-}
-
 template <typename I>
 bool ObjectMap<I>::update_required(const ceph::BitVector<2>::Iterator& it,
                                    uint8_t new_state) {

src/librbd/ObjectMap.h

@@ -65,7 +65,6 @@ class ObjectMap : public RefCountedObject {
   void close(Context *on_finish);
   bool set_object_map(ceph::BitVector<2> &target_object_map);
   bool object_may_exist(uint64_t object_no) const;
-  bool object_may_not_exist(uint64_t object_no) const;
 
   void aio_save(Context *on_finish);
   void aio_resize(uint64_t new_size, uint8_t default_object_state,

src/librbd/operation/FlattenRequest.cc

@@ -49,15 +49,6 @@ class C_FlattenObject : public C_AsyncObjectThrottle<I> {
       return -ERESTART;
     }
 
-    {
-      std::shared_lock image_lock{image_ctx.image_lock};
-      if (image_ctx.object_map != nullptr &&
-          !image_ctx.object_map->object_may_not_exist(m_object_no)) {
-        // can skip because the object already exists
-        return 1;
-      }
-    }
-
     if (!io::util::trigger_copyup(
             &image_ctx, m_object_no, m_io_context, this)) {
       // stop early if the parent went away - it just means

src/test/librbd/test_internal.cc

@@ -1571,6 +1571,83 @@ TEST_F(TestInternal, FlattenNoEmptyObjects)
   rados_ioctx_destroy(d_ioctx);
 }
 
+TEST_F(TestInternal, FlattenInconsistentObjectMap)
+{
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_OBJECT_MAP);
+  REQUIRE(!is_feature_enabled(RBD_FEATURE_STRIPINGV2));
+
+  librbd::ImageCtx* ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  librbd::NoOpProgressContext no_op;
+  ASSERT_EQ(0, ictx->operations->resize((1 << ictx->order) * 5, true, no_op));
+
+  bufferlist bl;
+  bl.append(std::string(256, '1'));
+  for (int i = 1; i < 5; i++) {
+    ASSERT_EQ(256, api::Io<>::write(*ictx, (1 << ictx->order) * i, 256,
+                                    bufferlist{bl}, 0));
+  }
+
+  ASSERT_EQ(0, snap_create(*ictx, "snap"));
+  ASSERT_EQ(0, snap_protect(*ictx, "snap"));
+
+  uint64_t features;
+  ASSERT_EQ(0, librbd::get_features(ictx, &features));
+
+  std::string clone_name = get_temp_image_name();
+  int order = ictx->order;
+  ASSERT_EQ(0, librbd::clone(m_ioctx, m_image_name.c_str(), "snap", m_ioctx,
+                             clone_name.c_str(), features, &order, 0, 0));
+
+  close_image(ictx);
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  C_SaferCond lock_ctx;
+  {
+    std::shared_lock owner_locker{ictx->owner_lock};
+    ictx->exclusive_lock->try_acquire_lock(&lock_ctx);
+  }
+  ASSERT_EQ(0, lock_ctx.wait());
+  ASSERT_TRUE(ictx->exclusive_lock->is_lock_owner());
+
+  ceph::BitVector<2> inconsistent_object_map;
+  inconsistent_object_map.resize(5);
+  inconsistent_object_map[0] = OBJECT_NONEXISTENT;
+  inconsistent_object_map[1] = OBJECT_NONEXISTENT;
+  inconsistent_object_map[2] = OBJECT_EXISTS;
+  inconsistent_object_map[3] = OBJECT_EXISTS_CLEAN;
+  // OBJECT_PENDING shouldn't happen within parent overlap, but test
+  // anyway
+  inconsistent_object_map[4] = OBJECT_PENDING;
+
+  auto object_map = new librbd::ObjectMap<>(*ictx, CEPH_NOSNAP);
+  C_SaferCond save_ctx;
+  {
+    std::shared_lock owner_locker{ictx->owner_lock};
+    std::unique_lock image_locker{ictx->image_lock};
+    object_map->set_object_map(inconsistent_object_map);
+    object_map->aio_save(&save_ctx);
+  }
+  ASSERT_EQ(0, save_ctx.wait());
+  object_map->put();
+
+  close_image(ictx);
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+  ASSERT_EQ(0, ictx->operations->flatten(no_op));
+
+  bufferptr read_ptr(256);
+  bufferlist read_bl;
+  read_bl.push_back(read_ptr);
+
+  librbd::io::ReadResult read_result{&read_bl};
+  for (int i = 1; i < 5; i++) {
+    ASSERT_EQ(256, api::Io<>::read(*ictx, (1 << ictx->order) * i, 256,
+                                   librbd::io::ReadResult{read_result}, 0));
+    EXPECT_TRUE(bl.contents_equal(read_bl));
+  }
+}
+
 TEST_F(TestInternal, PoolMetadataConfApply) {
   REQUIRE_FORMAT_V2();