test / librbd: create few empty objects during flatten

Fixes: http://tracker.ceph.com/issues/15028
Signed-off-by: Venky Shankar <vshankar@redhat.com>

src/test/librbd/test_librbd.cc

@@ -3369,6 +3369,138 @@ TEST_F(TestLibRBD, Flatten)
   ASSERT_PASSED(validate_object_map, clone_image);
 }
 
+TEST_F(TestLibRBD, FlattenNoEmptyObjects)
+{
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  rados_ioctx_t ioctx;
+  rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
+
+  librbd::RBD rbd;
+  std::string parent_name = get_temp_image_name();
+  uint64_t size = 4 << 20;
+  int order = 12; // smallest object size is 4K
+
+  bool old_format;
+  uint64_t features;
+  ASSERT_EQ(0, get_features(&old_format, &features));
+  ASSERT_FALSE(old_format);
+
+  // make a parent to clone from
+  ASSERT_EQ(0, create_image_full(ioctx, parent_name.c_str(), size, &order,
+                                 false, features));
+
+  rbd_image_t parent;
+  const int object_size = 1 << order;
+  const int object_num = size / object_size;
+  ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, NULL));
+  printf("made parent image \"%s\": %ldK (%d * %dK)\n", parent_name.c_str(),
+         (unsigned long)size, object_num, object_size/1024);
+
+  // write something into parent
+  char test_data[TEST_IO_SIZE + 1];
+  char zero_data[TEST_IO_SIZE + 1];
+  int i;
+  for (i = 0; i < TEST_IO_SIZE; ++i)
+    test_data[i] = (char) (rand() % (126 - 33) + 33);
+  test_data[TEST_IO_SIZE] = '\0';
+  memset(zero_data, 0, sizeof(zero_data));
+
+  // generate a random map which covers every objects with random
+  // offset
+  int count = 0;
+  map<uint64_t, uint64_t> write_tracker;
+  while (count < 10) {
+    uint64_t ono = rand() % object_num;
+    if (write_tracker.find(ono) == write_tracker.end()) {
+      uint64_t offset = rand() % (object_size - TEST_IO_SIZE);
+      write_tracker.insert(pair<uint64_t, uint64_t>(ono, offset));
+      count++;
+    }
+  }
+
+  printf("generated random write map:\n");
+  for (map<uint64_t, uint64_t>::iterator itr = write_tracker.begin();
+       itr != write_tracker.end(); ++itr)
+    printf("\t [%-8ld, %-8ld]\n",
+	   (unsigned long)itr->first, (unsigned long)itr->second);
+
+  printf("write data based on random map\n");
+  for (map<uint64_t, uint64_t>::iterator itr = write_tracker.begin();
+       itr != write_tracker.end(); ++itr) {
+    printf("\twrite object-%-4ld\t", (unsigned long)itr->first);
+    ASSERT_PASSED(write_test_data, parent, test_data, itr->first * object_size + itr->second, TEST_IO_SIZE, 0);
+  }
+
+  for (map<uint64_t, uint64_t>::iterator itr = write_tracker.begin();
+         itr != write_tracker.end(); ++itr) {
+    printf("\tread object-%-4ld\t", (unsigned long)itr->first);
+    ASSERT_PASSED(read_test_data, parent, test_data, itr->first * object_size + itr->second, TEST_IO_SIZE, 0);
+  }
+
+  // find out what objects the parent image has generated
+  rbd_image_info_t p_info;
+  ASSERT_EQ(0, rbd_stat(parent, &p_info, sizeof(p_info)));
+
+  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);
+
+  const char *entry;
+  set<string> obj_checker;
+  rados_list_ctx_t 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, p_info.block_name_prefix)) {
+      const char *block_name_suffix = entry + strlen(p_info.block_name_prefix) + 1;
+      obj_checker.insert(block_name_suffix);
+    }
+  }
+  rados_nobjects_list_close(list_ctx);
+  ASSERT_EQ(obj_checker.size(), write_tracker.size());
+
+  // create a snapshot, reopen as the parent we're interested in and protect it
+  ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
+  ASSERT_EQ(0, rbd_close(parent));
+  ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, "parent_snap"));
+  ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
+  ASSERT_PASSED(validate_object_map, parent);
+  ASSERT_EQ(0, rbd_close(parent));
+  printf("made snapshot and protected: \"%s@parent_snap\"\n", parent_name.c_str());
+
+  std::string child_name = get_temp_image_name();
+  // create a copy-on-read clone and open it
+  ASSERT_EQ(0, rbd_clone(ioctx, parent_name.c_str(), "parent_snap", ioctx,
+                         child_name.c_str(), features, &order));
+
+  rbd_image_t child;
+  ASSERT_EQ(0, rbd_open(ioctx, child_name.c_str(), &child, NULL));
+  printf("made and opened clone \"%s\"\n", child_name.c_str());
+
+  printf("flattening clone: \"%s\"\n", child_name.c_str());
+  ASSERT_EQ(0, rbd_flatten(child));
+
+  printf("check whether child image has the same set of objects as parent\n");
+  rbd_image_info_t c_info;
+  ASSERT_EQ(0, rbd_stat(child, &c_info, sizeof(c_info)));
+  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;
+      set<string>::iterator it = obj_checker.find(block_name_suffix);
+      ASSERT_TRUE(it != obj_checker.end());
+      obj_checker.erase(it);
+    }
+  }
+  rados_nobjects_list_close(list_ctx);
+  ASSERT_TRUE(obj_checker.empty());
+  ASSERT_PASSED(validate_object_map, child);
+  ASSERT_EQ(0, rbd_close(child));
+
+  rados_ioctx_destroy(ioctx);
+  rados_ioctx_destroy(d_ioctx);
+}
+
 TEST_F(TestLibRBD, SnapshotLimit)
 {
   rados_ioctx_t ioctx;