librbd: bug fixes for optional data pool support

qa/workunits/rbd/import_export.sh

@@ -1,5 +1,16 @@
 #!/bin/sh -ex
 
+# returns data pool for a given image
+get_image_data_pool () {
+    image=$1
+    data_pool=$(rbd info $image | grep "data_pool: " | awk -F':' '{ print $NF }')
+    if [ -z $data_pool ]; then
+       data_pool='rbd'
+    fi
+
+    echo $data_pool
+}
+
 # return list of object numbers populated in image
 objects () {
    image=$1
@@ -8,7 +19,7 @@ objects () {
    # strip off prefix and leading zeros from objects; sort, although
    # it doesn't necessarily make sense as they're hex, at least it makes
    # the list repeatable and comparable
-   objects=$(rados ls -p rbd | grep $prefix | \
+   objects=$(rados ls -p $(get_image_data_pool $image) | grep $prefix | \
        sed -e 's/'$prefix'\.//' -e 's/^0*\([0-9a-f]\)/\1/' | sort -u)
    echo $objects
 }
@@ -140,7 +151,8 @@ echo "zeros export to sparse file"
 rbd create $RBD_CREATE_ARGS sparse --size 4
 prefix=$(rbd info sparse | grep block_name_prefix | awk '{print $NF;}')
 # drop in 0 object directly
-dd if=/dev/zero bs=4M count=1 | rados -p rbd put ${prefix}.000000000000 -
+dd if=/dev/zero bs=4M count=1 | rados -p $(get_image_data_pool sparse) \
+                                      put ${prefix}.000000000000 -
 [ $tiered -eq 1 -o "$(objects sparse)" = '0' ]
 # 1 object full of zeros; export should still create 0-disk-usage file
 rm ${TMPDIR}/sparse || true

src/librados/librados.cc

@@ -2413,6 +2413,7 @@ int librados::Rados::ioctx_create(const char *name, IoCtx &io)
   int ret = rados_ioctx_create((rados_t)client, name, &p);
   if (ret)
     return ret;
+  io.close();
   io.io_ctx_impl = (IoCtxImpl*)p;
   return 0;
 }
@@ -2423,6 +2424,7 @@ int librados::Rados::ioctx_create2(int64_t pool_id, IoCtx &io)
   int ret = rados_ioctx_create2((rados_t)client, pool_id, &p);
   if (ret)
     return ret;
+  io.close();
   io.io_ctx_impl = (IoCtxImpl*)p;
   return 0;
 }

src/librbd/CopyupRequest.cc

@@ -146,7 +146,12 @@ bool CopyupRequest::send_copyup() {
     ldout(m_ictx->cct, 20) << __func__ << " " << this << " copyup with "
                            << "empty snapshot context" << dendl;
     librados::AioCompletion *comp = util::create_rados_safe_callback(this);
-    r = m_ictx->md_ctx.aio_operate(m_oid, comp, &copyup_op, 0, snaps);
+
+    librados::Rados rados(m_ictx->data_ctx);
+    r = rados.ioctx_create2(m_ictx->data_ctx.get_id(), m_data_ctx);
+    assert(r == 0);
+
+    r = m_data_ctx.aio_operate(m_oid, comp, &copyup_op, 0, snaps);
     assert(r == 0);
     comp->release();
   }

src/librbd/CopyupRequest.h

@@ -74,6 +74,7 @@ class CopyupRequest {
   AsyncOperation m_async_op;
 
   std::vector<uint64_t> m_snap_ids;
+  librados::IoCtx m_data_ctx; // for empty SnapContext
 
   void complete_requests(int r);
 

src/librbd/operation/ObjectMapIterate.cc

@@ -38,7 +38,7 @@ class C_VerifyObjectCallback : public C_AsyncObjectThrottle<I> {
     m_handle_mismatch(handle_mismatch),
     m_invalidate(invalidate)
   {
-    m_io_ctx.dup(image_ctx->md_ctx);
+    m_io_ctx.dup(image_ctx->data_ctx);
     m_io_ctx.snap_set_read(CEPH_SNAPDIR);
   }
 

src/test/librados_test_stub/LibradosTestStub.cc

@@ -906,6 +906,8 @@ int Rados::ioctx_create(const char *name, IoCtx &io) {
   if (ret) {
     return ret;
   }
+
+  io.close();
   io.io_ctx_impl = reinterpret_cast<IoCtxImpl*>(p);
   return 0;
 }
@@ -917,6 +919,8 @@ int Rados::ioctx_create2(int64_t pool_id, IoCtx &io)
   if (ret) {
     return ret;
   }
+
+  io.close();
   io.io_ctx_impl = reinterpret_cast<IoCtxImpl*>(p);
   return 0;
 }

src/test/librbd/test_ObjectMap.cc

@@ -41,7 +41,7 @@ TEST_F(TestObjectMap, RefreshInvalidatesWhenCorrupt) {
   std::string oid = librbd::ObjectMap::object_map_name(ictx->id, CEPH_NOSNAP);
   bufferlist bl;
   bl.append("corrupt");
-  ASSERT_EQ(0, ictx->data_ctx.write_full(oid, bl));
+  ASSERT_EQ(0, ictx->md_ctx.write_full(oid, bl));
 
   ASSERT_EQ(0, when_open_object_map(ictx));
   ASSERT_TRUE(ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID));
@@ -65,7 +65,7 @@ TEST_F(TestObjectMap, RefreshInvalidatesWhenTooSmall) {
   librbd::cls_client::object_map_resize(&op, 0, OBJECT_NONEXISTENT);
 
   std::string oid = librbd::ObjectMap::object_map_name(ictx->id, CEPH_NOSNAP);
-  ASSERT_EQ(0, ictx->data_ctx.operate(oid, &op));
+  ASSERT_EQ(0, ictx->md_ctx.operate(oid, &op));
 
   ASSERT_EQ(0, when_open_object_map(ictx));
   ASSERT_TRUE(ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID));
@@ -88,7 +88,7 @@ TEST_F(TestObjectMap, InvalidateFlagOnDisk) {
   std::string oid = librbd::ObjectMap::object_map_name(ictx->id, CEPH_NOSNAP);
   bufferlist bl;
   bl.append("corrupt");
-  ASSERT_EQ(0, ictx->data_ctx.write_full(oid, bl));
+  ASSERT_EQ(0, ictx->md_ctx.write_full(oid, bl));
 
   ASSERT_EQ(0, when_open_object_map(ictx));
   ASSERT_TRUE(ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID));
@@ -106,16 +106,16 @@ TEST_F(TestObjectMap, InvalidateFlagInMemoryOnly) {
 
   std::string oid = librbd::ObjectMap::object_map_name(ictx->id, CEPH_NOSNAP);
   bufferlist valid_bl;
-  ASSERT_LT(0, ictx->data_ctx.read(oid, valid_bl, 0, 0));
+  ASSERT_LT(0, ictx->md_ctx.read(oid, valid_bl, 0, 0));
 
   bufferlist corrupt_bl;
   corrupt_bl.append("corrupt");
-  ASSERT_EQ(0, ictx->data_ctx.write_full(oid, corrupt_bl));
+  ASSERT_EQ(0, ictx->md_ctx.write_full(oid, corrupt_bl));
 
   ASSERT_EQ(0, when_open_object_map(ictx));
   ASSERT_TRUE(ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID));
 
-  ASSERT_EQ(0, ictx->data_ctx.write_full(oid, valid_bl));
+  ASSERT_EQ(0, ictx->md_ctx.write_full(oid, valid_bl));
   ASSERT_EQ(0, open_image(m_image_name, &ictx));
   ASSERT_FALSE(ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID));
 }

src/test/librbd/test_fixture.cc

@@ -20,16 +20,30 @@
 std::string TestFixture::_pool_name;
 librados::Rados TestFixture::_rados;
 uint64_t TestFixture::_image_number = 0;
+std::string TestFixture::_data_pool;
 
 TestFixture::TestFixture() : m_image_size(0) {
 }
 
 void TestFixture::SetUpTestCase() {
   _pool_name = get_temp_pool_name("test-librbd-");
   ASSERT_EQ("", create_one_pool_pp(_pool_name, _rados));
+
+  bool created = false;
+  ASSERT_EQ(0, create_image_data_pool(_rados, _data_pool, &created));
+  if (!_data_pool.empty()) {
+    printf("using image data pool: %s\n", _data_pool.c_str());
+    if (!created) {
+      _data_pool.clear();
+    }
+  }
 }
 
 void TestFixture::TearDownTestCase() {
+  if (!_data_pool.empty()) {
+    ASSERT_EQ(0, _rados.pool_delete(_data_pool.c_str()));
+  }
+
   ASSERT_EQ(0, destroy_one_pool_pp(_pool_name, _rados));
 }
 

src/test/librbd/test_fixture.h

@@ -40,6 +40,7 @@ class TestFixture : public ::testing::Test {
   static std::string _pool_name;
   static librados::Rados _rados;
   static uint64_t _image_number;
+  static std::string _data_pool;
 
   librados::IoCtx m_ioctx;
   librbd::RBD m_rbd;

src/test/librbd/test_internal.cc

@@ -515,7 +515,7 @@ TEST_F(TestInternal, SnapshotCopyup)
   ASSERT_EQ(256, ictx2->aio_work_queue->write(256, bl.length(), bl.c_str(), 0));
 
   librados::IoCtx snap_ctx;
-  snap_ctx.dup(m_ioctx);
+  snap_ctx.dup(ictx2->data_ctx);
   snap_ctx.snap_set_read(CEPH_SNAPDIR);
 
   librados::snap_set_t snap_set;

src/test/librbd/test_librbd.cc

@@ -161,6 +161,8 @@ class TestLibRBD : public ::testing::Test {
     _image_number = 0;
     ASSERT_EQ("", connect_cluster(&_cluster));
     ASSERT_EQ("", connect_cluster_pp(_rados));
+
+    create_optional_data_pool();
   }
 
   static void TearDownTestCase() {
@@ -197,6 +199,18 @@ class TestLibRBD : public ::testing::Test {
     return "image" + stringify(_image_number);
   }
 
+  static void create_optional_data_pool() {
+    bool created = false;
+    std::string data_pool;
+    ASSERT_EQ(0, create_image_data_pool(_rados, data_pool, &created));
+    if (!data_pool.empty()) {
+      printf("using image data pool: %s\n", data_pool.c_str());
+      if (created) {
+        _unique_pool_names.push_back(data_pool);
+      }
+    }
+  }
+
   std::string create_pool(bool unique = false) {
     librados::Rados rados;
     std::string pool_name;
@@ -2146,7 +2160,12 @@ TEST_F(TestLibRBD, TestCoR)
 
   // find out what objects the parent image has generated
   ASSERT_EQ(0, rbd_stat(parent, &p_info, sizeof(p_info)));
-  ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &list_ctx));
+
+  int64_t data_pool_id = rbd_get_data_pool_id(parent);
+  rados_ioctx_t d_ioctx;
+  rados_ioctx_create2(_cluster, data_pool_id, &d_ioctx);
+
+  ASSERT_EQ(0, rados_nobjects_list_open(d_ioctx, &list_ctx));
   while (rados_nobjects_list_next(list_ctx, &entry, NULL, NULL) != -ENOENT) {
     if (strstr(entry, p_info.block_name_prefix)) {
       const char *block_name_suffix = entry + strlen(p_info.block_name_prefix) + 1;
@@ -2197,7 +2216,7 @@ TEST_F(TestLibRBD, TestCoR)
   printf("check whether child image has the same set of objects as parent\n");
   ASSERT_EQ(0, rbd_open(ioctx, "child", &child, NULL));
   ASSERT_EQ(0, rbd_stat(child, &c_info, sizeof(c_info)));
-  ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &list_ctx));
+  ASSERT_EQ(0, rados_nobjects_list_open(d_ioctx, &list_ctx));
   while (rados_nobjects_list_next(list_ctx, &entry, NULL, NULL) != -ENOENT) {
     if (strstr(entry, c_info.block_name_prefix)) {
       const char *block_name_suffix = entry + strlen(c_info.block_name_prefix) + 1;
@@ -2212,6 +2231,7 @@ TEST_F(TestLibRBD, TestCoR)
   ASSERT_EQ(0, rbd_close(child));
 
   rados_ioctx_destroy(ioctx);
+  rados_ioctx_destroy(d_ioctx);
 }
 
 static void test_list_children(rbd_image_t image, ssize_t num_expected, ...)

src/test/librbd/test_support.cc

@@ -48,3 +48,21 @@ int get_image_id(librbd::Image &image, std::string *image_id)
   return 0;
 }
 
+int create_image_data_pool(librados::Rados &rados, std::string &data_pool, bool *created) {
+  std::string pool;
+  int r = rados.conf_get("rbd_default_data_pool", pool);
+  if (r != 0) {
+    return r;
+  } else if (pool.empty()) {
+    return 0;
+  }
+
+  r = rados.pool_create(pool.c_str());
+  if ((r == 0) || (r == -EEXIST)) {
+    data_pool = pool;
+    *created = (r == 0);
+    return 0;
+  }
+
+  return r;
+}

src/test/librbd/test_support.h

@@ -10,6 +10,7 @@ bool is_feature_enabled(uint64_t feature);
 int create_image_pp(librbd::RBD &rbd, librados::IoCtx &ioctx,
                     const std::string &name, uint64_t size);
 int get_image_id(librbd::Image &image, std::string *image_id);
+int create_image_data_pool(librados::Rados &rados, std::string &data_pool, bool *created);
 
 #define REQUIRE(x) {			  \
   if (!(x)) {				  \

src/test/pybind/test_rbd.py

@@ -133,6 +133,10 @@ def check_default_params(format, order=None, features=None, stripe_count=None,
             rados.conf_set('rbd_default_stripe_count', str(stripe_count or 0))
         if stripe_unit is not None:
             rados.conf_set('rbd_default_stripe_unit', str(stripe_unit or 0))
+        feature_data_pool = 0
+        datapool = rados.conf_get('rbd_default_data_pool')
+        if not len(datapool) == 0:
+            feature_data_pool = 128
         image_name = get_temp_image_name()
         if exception is None:
             RBD().create(ioctx, image_name, IMG_SIZE)
@@ -145,8 +149,12 @@ def check_default_params(format, order=None, features=None, stripe_count=None,
                     eq(expected_order, actual_order)
 
                     expected_features = features
-                    if expected_features is None or format == 1:
-                        expected_features = 0 if format == 1 else 61
+                    if format == 1:
+                        expected_features = 0
+                    elif expected_features is None:
+                        expected_features = 61 | feature_data_pool
+                    else:
+                        expected_features |= feature_data_pool
                     eq(expected_features, image.features())
 
                     expected_stripe_count = stripe_count

src/test/rbd_mirror/test_ClusterWatcher.cc

@@ -7,6 +7,7 @@
 #include "librbd/internal.h"
 #include "tools/rbd_mirror/ClusterWatcher.h"
 #include "tools/rbd_mirror/types.h"
+#include "test/rbd_mirror/test_fixture.h"
 #include "test/librados/test.h"
 #include "gtest/gtest.h"
 #include <boost/scope_exit.hpp>
@@ -25,7 +26,7 @@ using std::string;
 void register_test_cluster_watcher() {
 }
 
-class TestClusterWatcher : public ::testing::Test {
+class TestClusterWatcher : public ::rbd::mirror::TestFixture {
 public:
 
   TestClusterWatcher() : m_lock("TestClusterWatcherLock")

src/test/rbd_mirror/test_ImageReplayer.cc

@@ -17,6 +17,7 @@
 #include "include/rados/librados.hpp"
 #include "include/rbd/librbd.hpp"
 #include "include/stringify.h"
+#include "test/rbd_mirror/test_fixture.h"
 #include "cls/journal/cls_journal_types.h"
 #include "cls/journal/cls_journal_client.h"
 #include "cls/rbd/cls_rbd_types.h"
@@ -48,7 +49,7 @@ void register_test_rbd_mirror() {
 #define TEST_IO_SIZE 512
 #define TEST_IO_COUNT 11
 
-class TestImageReplayer : public ::testing::Test {
+class TestImageReplayer : public ::rbd::mirror::TestFixture {
 public:
   struct C_WatchCtx : public librados::WatchCtx2 {
     TestImageReplayer *test;

src/test/rbd_mirror/test_PoolWatcher.cc

@@ -3,6 +3,7 @@
 #include "include/rados/librados.hpp"
 #include "include/rbd/librbd.hpp"
 #include "include/stringify.h"
+#include "test/rbd_mirror/test_fixture.h"
 #include "cls/rbd/cls_rbd_types.h"
 #include "cls/rbd/cls_rbd_client.h"
 #include "include/rbd_types.h"
@@ -35,7 +36,7 @@ using std::string;
 void register_test_pool_watcher() {
 }
 
-class TestPoolWatcher : public ::testing::Test {
+class TestPoolWatcher : public ::rbd::mirror::TestFixture {
 public:
 
 TestPoolWatcher() : m_lock("TestPoolWatcherLock"),