test: fix TierFlushDuringFlush to wait until dedup_tier is set on base pool

When start_dedup() is called while the base pool is not set the dedup_tier,
it is not possible to know the target pool of the chunk object.

1. User set the dedup_tier on a base pool by mon_command().
2. User issues tier_flush on the object which has a manifest (base pool)
  before the dedup_tier is applied on the base pool.
3. OSD calls start_dedup() to flush the chunk objects to chunk pool.
4. OSD calls get_dedup_tier() to get the chunk pool of the base pool,
  but it is not possible to know the chunk pool.
5. get_dedup_tier() returns 0 because it is not applied on the base pool yet.
6. This makes refcount_manifest() lost it's way to chunk pool.

To prevent this issue, start_dedup() has to be called after dedup_tier is set
on the base pool. To do so, this commit prohibits getting chunk pool id if
dedup_tier is not set.

Fixes: http://tracker.ceph.com/issues/53855
Signed-off-by: Sungmin Lee <sung_min.lee@samsung.com>

diff --git a/src/osd/PrimaryLogPG.cc b/src/osd/PrimaryLogPG.cc
index e91213bdee92168004feae70291f62581397b240..b245e86b2272b5a2efbeea8707393a330cc5c2d2 100644 (file)
--- a/src/osd/PrimaryLogPG.cc
+++ b/src/osd/PrimaryLogPG.cc
@@ -10457,7 +10457,11 @@ int PrimaryLogPG::start_dedup(OpRequestRef op, ObjectContextRef obc)
   if (pool.info.get_fingerprint_type() == pg_pool_t::TYPE_FINGERPRINT_NONE) {
     dout(0) << " fingerprint algorithm is not set " << dendl;
     return -EINVAL;
-  } 
+  }
+  if (pool.info.get_dedup_tier() <= 0) {
+    dout(10) << " dedup tier is not set " << dendl;
+    return -EINVAL;
+  }
 
   /*
    * The operations to make dedup chunks are tracked by a ManifestOp.
@@ -10593,6 +10597,8 @@ hobject_t PrimaryLogPG::get_fpoid_from_chunk(const hobject_t soid, bufferlist& c
   pg_t raw_pg;
   object_locator_t oloc(soid);
   oloc.pool = pool.info.get_dedup_tier();
+  // check if dedup_tier isn't set
+  ceph_assert(oloc.pool > 0);
   get_osdmap()->object_locator_to_pg(fp_oid, oloc, raw_pg);
   hobject_t target(fp_oid, oloc.key, snapid_t(),
                    raw_pg.ps(), raw_pg.pool(),

diff --git a/src/test/librados/tier_cxx.cc b/src/test/librados/tier_cxx.cc
index 287c6fa2ae6ac5d396296b83f7ab3d671d1e1456..f34accf8f1340612db5e5ca7c176356c5ae509e9 100644 (file)
--- a/src/test/librados/tier_cxx.cc
+++ b/src/test/librados/tier_cxx.cc
@@ -5659,6 +5659,9 @@ TEST_F(LibRadosTwoPoolsPP, TierFlushDuringFlush) {
     ASSERT_EQ(0, ioctx.operate("bar", &op));
   }
 
+  // wait for maps to settle
+  cluster.wait_for_latest_osdmap();
+
   // set-chunk to set manifest object
   {
     ObjectReadOperation op;
@@ -9143,3 +9146,68 @@ TEST_F(LibRadosTwoPoolsPP, HelloWriteReturn) {
     ASSERT_EQ("", std::string(out.c_str(), out.length()));
   }
 }
+
+TEST_F(LibRadosTwoPoolsPP, TierFlushDuringUnsetDedupTier) {
+  // skip test if not yet octopus
+  if (_get_required_osd_release(cluster) < "octopus") {
+    cout << "cluster is not yet octopus, skipping test" << std::endl;
+    return;
+  }
+
+  bufferlist inbl;
+
+  // set dedup parameters without dedup_tier
+  ASSERT_EQ(0, cluster.mon_command(
+    set_pool_str(cache_pool_name, "fingerprint_algorithm", "sha1"),
+    inbl, NULL, NULL));
+  ASSERT_EQ(0, cluster.mon_command(
+    set_pool_str(cache_pool_name, "dedup_chunk_algorithm", "fastcdc"),
+    inbl, NULL, NULL));
+  ASSERT_EQ(0, cluster.mon_command(
+    set_pool_str(cache_pool_name, "dedup_cdc_chunk_size", 1024),
+    inbl, NULL, NULL));
+
+  // create object
+  bufferlist gbl;
+  {
+    generate_buffer(1024*8, &gbl);
+    ObjectWriteOperation op;
+    op.write_full(gbl);
+    ASSERT_EQ(0, cache_ioctx.operate("foo", &op));
+  }
+  {
+    bufferlist bl;
+    bl.append("there hiHI");
+    ObjectWriteOperation op;
+    op.write_full(bl);
+    ASSERT_EQ(0, ioctx.operate("bar", &op));
+  }
+
+  // wait for maps to settle
+  cluster.wait_for_latest_osdmap();
+
+  // set-chunk to set manifest object
+  {
+    ObjectReadOperation op;
+    op.set_chunk(0, 2, ioctx, "bar", 0, CEPH_OSD_OP_FLAG_WITH_REFERENCE);
+    librados::AioCompletion *completion = cluster.aio_create_completion();
+    ASSERT_EQ(0, cache_ioctx.aio_operate("foo", completion, &op,
+      librados::OPERATION_IGNORE_CACHE, NULL));
+    completion->wait_for_complete();
+    ASSERT_EQ(0, completion->get_return_value());
+    completion->release();
+  }
+
+  // flush to check if proper error is returned
+  {
+    ObjectReadOperation op;
+    op.tier_flush();
+    librados::AioCompletion *completion = cluster.aio_create_completion();
+    ASSERT_EQ(0, cache_ioctx.aio_operate(
+      "foo", completion, &op, librados::OPERATION_IGNORE_CACHE, NULL));
+    completion->wait_for_complete();
+    ASSERT_EQ(-EINVAL, completion->get_return_value());
+    completion->release();
+  }
+}
+