image_sync/test_mock_SnapshotCreateRequest.cc
image_sync/test_mock_SyncPointCreateRequest.cc
image_sync/test_mock_SyncPointPruneRequest.cc
+ pool_watcher/test_mock_RefreshImagesRequest.cc
)
add_ceph_unittest(unittest_rbd_mirror ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/unittest_rbd_mirror)
set_target_properties(unittest_rbd_mirror PROPERTIES COMPILE_FLAGS
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/rbd_mirror/test_mock_fixture.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "tools/rbd_mirror/pool_watcher/RefreshImagesRequest.h"
+#include "include/stringify.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public librbd::MockImageCtx {
+ MockTestImageCtx(librbd::ImageCtx &image_ctx)
+ : librbd::MockImageCtx(image_ctx) {
+ }
+};
+
+} // anonymous namespace
+} // namespace librbd
+
+// template definitions
+#include "tools/rbd_mirror/pool_watcher/RefreshImagesRequest.cc"
+template class rbd::mirror::pool_watcher::RefreshImagesRequest<librbd::MockTestImageCtx>;
+
+namespace rbd {
+namespace mirror {
+namespace pool_watcher {
+
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::StrEq;
+using ::testing::WithArg;
+
+class TestMockPoolWatcherRefreshImagesRequest : public TestMockFixture {
+public:
+ typedef RefreshImagesRequest<librbd::MockTestImageCtx> MockRefreshImagesRequest;
+
+ void expect_mirror_image_list(librados::IoCtx &io_ctx,
+ const std::map<std::string, std::string> &ids,
+ int r) {
+ bufferlist bl;
+ ::encode(ids, bl);
+
+ EXPECT_CALL(get_mock_io_ctx(io_ctx),
+ exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_image_list"), _, _, _))
+ .WillOnce(DoAll(WithArg<5>(Invoke([bl](bufferlist *out_bl) {
+ *out_bl = bl;
+ })),
+ Return(r)));
+ }
+
+ void expect_dir_list(librados::IoCtx &io_ctx,
+ const std::map<std::string, std::string> &ids, int r) {
+ bufferlist bl;
+ ::encode(ids, bl);
+
+ EXPECT_CALL(get_mock_io_ctx(io_ctx),
+ exec(RBD_DIRECTORY, _, StrEq("rbd"), StrEq("dir_list"), _, _, _))
+ .WillOnce(DoAll(WithArg<5>(Invoke([bl](bufferlist *out_bl) {
+ *out_bl = bl;
+ })),
+ Return(r)));
+ }
+};
+
+TEST_F(TestMockPoolWatcherRefreshImagesRequest, Success) {
+ InSequence seq;
+ expect_mirror_image_list(m_remote_io_ctx, {{"local id", "global id"}}, 0);
+ expect_dir_list(m_remote_io_ctx, {{"image name", "local id"}}, 0);
+
+ C_SaferCond ctx;
+ ImageIds image_ids;
+ MockRefreshImagesRequest *req = new MockRefreshImagesRequest(
+ m_remote_io_ctx, &image_ids, &ctx);
+
+ req->send();
+ ASSERT_EQ(0, ctx.wait());
+
+ ImageIds expected_image_ids = {{"global id", "local id",
+ boost::optional<std::string>{"image name"}}};
+ ASSERT_EQ(expected_image_ids, image_ids);
+}
+
+TEST_F(TestMockPoolWatcherRefreshImagesRequest, LargeDirectory) {
+ InSequence seq;
+ std::map<std::string, std::string> mirror_list;
+ std::map<std::string, std::string> dir_list;
+ ImageIds expected_image_ids;
+ for (uint32_t idx = 1; idx <= 1024; ++idx) {
+ mirror_list.insert(std::make_pair("local id " + stringify(idx),
+ "global id " + stringify(idx)));
+ dir_list.insert(std::make_pair("image " + stringify(idx),
+ "local id " + stringify(idx)));
+ expected_image_ids.insert({{"global id " + stringify(idx),
+ "local id " + stringify(idx),
+ "image " + stringify(idx)}});
+ }
+
+ expect_mirror_image_list(m_remote_io_ctx, mirror_list, 0);
+ expect_mirror_image_list(m_remote_io_ctx, {{"local id", "global id"}}, 0);
+ expect_dir_list(m_remote_io_ctx, dir_list, 0);
+ expect_dir_list(m_remote_io_ctx, {{"image name", "local id"}}, 0);
+
+ C_SaferCond ctx;
+ ImageIds image_ids;
+ MockRefreshImagesRequest *req = new MockRefreshImagesRequest(
+ m_remote_io_ctx, &image_ids, &ctx);
+
+ req->send();
+ ASSERT_EQ(0, ctx.wait());
+
+ expected_image_ids.insert({"global id", "local id",
+ boost::optional<std::string>{"image name"}});
+ ASSERT_EQ(expected_image_ids, image_ids);
+}
+
+TEST_F(TestMockPoolWatcherRefreshImagesRequest, MirrorImageListError) {
+ InSequence seq;
+ expect_mirror_image_list(m_remote_io_ctx, {}, -EINVAL);
+
+ C_SaferCond ctx;
+ ImageIds image_ids;
+ MockRefreshImagesRequest *req = new MockRefreshImagesRequest(
+ m_remote_io_ctx, &image_ids, &ctx);
+
+ req->send();
+ ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockPoolWatcherRefreshImagesRequest, DirListError) {
+ InSequence seq;
+ expect_mirror_image_list(m_remote_io_ctx, {{"local id", "global id"}}, 0);
+ expect_dir_list(m_remote_io_ctx, {{"image name", "local id"}}, -EINVAL);
+
+ C_SaferCond ctx;
+ ImageIds image_ids;
+ MockRefreshImagesRequest *req = new MockRefreshImagesRequest(
+ m_remote_io_ctx, &image_ids, &ctx);
+
+ req->send();
+ ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+} // namespace pool_watcher
+} // namespace mirror
+} // namespace rbd
#include <set>
#include <vector>
+using rbd::mirror::ImageId;
+using rbd::mirror::ImageIds;
using rbd::mirror::PoolWatcher;
using rbd::mirror::peer_t;
using rbd::mirror::RadosRef;
sizeof(mirror_image_info)));
image.close();
- m_mirrored_images.insert(PoolWatcher::ImageId(
+ m_mirrored_images.insert(ImageId(
mirror_image_info.global_id, get_image_id(&ioctx, name), name));
}
if (image_name != nullptr)
sizeof(mirror_image_info)));
image.close();
- m_mirrored_images.insert(PoolWatcher::ImageId(
+ m_mirrored_images.insert(ImageId(
mirror_image_info.global_id, get_image_id(&cioctx, name), name));
}
if (image_name != nullptr)
unique_ptr<PoolWatcher> m_pool_watcher;
set<string> m_pools;
- PoolWatcher::ImageIds m_mirrored_images;
+ ImageIds m_mirrored_images;
uint64_t m_image_number;
uint64_t m_snap_number;
image_sync/SnapshotCopyRequest.cc
image_sync/SnapshotCreateRequest.cc
image_sync/SyncPointCreateRequest.cc
- image_sync/SyncPointPruneRequest.cc)
+ image_sync/SyncPointPruneRequest.cc
+ pool_watcher/RefreshImagesRequest.cc)
add_library(rbd_mirror_internal STATIC
${rbd_mirror_internal})
return m_blacklisted;
}
-const PoolWatcher::ImageIds& PoolWatcher::get_images() const
+const ImageIds& PoolWatcher::get_images() const
{
assert(m_lock.is_locked());
return m_images;
return r;
}
for (auto it = mirror_images.begin(); it != mirror_images.end(); ++it) {
- boost::optional<std::string> image_name(boost::none);
+ std::string image_name;
auto it2 = image_id_to_name.find(it->first);
if (it2 != image_id_to_name.end()) {
image_name = it2->second;
*/
class PoolWatcher {
public:
- struct ImageId {
- std::string global_id;
- std::string id;
- boost::optional<std::string> name;
-
- ImageId(const std::string &global_id, const std::string &id = "",
- const boost::optional<std::string> &name = boost::none)
- : global_id(global_id), id(id), name(name) {
- }
-
- inline bool operator==(const ImageId &rhs) const {
- return (global_id == rhs.global_id && id == rhs.id && name == rhs.name);
- }
- inline bool operator<(const ImageId &rhs) const {
- return global_id < rhs.global_id;
- }
- };
- typedef std::set<ImageId> ImageIds;
-
PoolWatcher(librados::IoCtx &remote_io_ctx, double interval_seconds,
Mutex &lock, Cond &cond);
~PoolWatcher();
// shut down replayers for non-mirrored images
for (auto image_it = m_image_replayers.begin();
image_it != m_image_replayers.end();) {
- auto image_id_it = image_ids.find(image_it->first);
+ auto image_id_it = image_ids.find(ImageId(image_it->first));
if (image_id_it == image_ids.end()) {
if (image_it->second->is_running()) {
dout(20) << "stop image replayer for remote image "
void release_leader();
private:
- typedef PoolWatcher::ImageId ImageId;
- typedef PoolWatcher::ImageIds ImageIds;
-
void init_local_mirroring_images();
void set_sources(const ImageIds &image_ids);
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "tools/rbd_mirror/pool_watcher/RefreshImagesRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/Utils.h"
+#include <map>
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rbd_mirror
+#undef dout_prefix
+#define dout_prefix *_dout << "rbd::mirror::pool_watcher::RefreshImagesRequest " \
+ << this << " " << __func__ << ": "
+
+namespace rbd {
+namespace mirror {
+namespace pool_watcher {
+
+static const uint32_t MAX_RETURN = 1024;
+
+using librbd::util::create_rados_ack_callback;
+
+template <typename I>
+void RefreshImagesRequest<I>::send() {
+ mirror_image_list();
+}
+
+template <typename I>
+void RefreshImagesRequest<I>::mirror_image_list() {
+ dout(10) << dendl;
+
+ librados::ObjectReadOperation op;
+ librbd::cls_client::mirror_image_list_start(&op, m_start_after, MAX_RETURN);
+
+ librados::AioCompletion *aio_comp = create_rados_ack_callback<
+ RefreshImagesRequest<I>,
+ &RefreshImagesRequest<I>::handle_mirror_image_list>(this);
+ int r = m_remote_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
+ assert(r == 0);
+ aio_comp->release();
+}
+
+template <typename I>
+void RefreshImagesRequest<I>::handle_mirror_image_list(int r) {
+ dout(10) << "r=" << r << dendl;
+
+ std::map<std::string, std::string> ids;
+ if (r == 0) {
+ bufferlist::iterator it = m_out_bl.begin();
+ r = librbd::cls_client::mirror_image_list_finish(&it, &ids);
+ }
+
+ if (r < 0) {
+ derr << "failed to list mirrored images: " << cpp_strerror(r) << dendl;
+ finish(r);
+ return;
+ }
+
+ if (!ids.empty()) {
+ m_local_to_global_ids.insert(ids.begin(), ids.end());
+ if (ids.size() == MAX_RETURN) {
+ m_start_after = ids.rbegin()->first;
+ mirror_image_list();
+ return;
+ }
+ }
+
+ dir_list();
+}
+
+template <typename I>
+void RefreshImagesRequest<I>::dir_list() {
+ dout(10) << dendl;
+
+ m_out_bl.clear();
+ m_start_after = "";
+
+ librados::ObjectReadOperation op;
+ librbd::cls_client::dir_list_start(&op, m_start_after, MAX_RETURN);
+
+ librados::AioCompletion *aio_comp = create_rados_ack_callback<
+ RefreshImagesRequest<I>,
+ &RefreshImagesRequest<I>::handle_dir_list>(this);
+ int r = m_remote_io_ctx.aio_operate(RBD_DIRECTORY, aio_comp, &op, &m_out_bl);
+ assert(r == 0);
+ aio_comp->release();
+}
+
+template <typename I>
+void RefreshImagesRequest<I>::handle_dir_list(int r) {
+ dout(10) << "r=" << r << dendl;
+
+ std::map<std::string, std::string> name_to_ids;
+ if (r == 0) {
+ bufferlist::iterator it = m_out_bl.begin();
+ r = librbd::cls_client::dir_list_finish(&it, &name_to_ids);
+ }
+
+ if (r < 0) {
+ derr << "failed to list images: " << cpp_strerror(r) << dendl;
+ finish(r);
+ return;
+ }
+
+ if (!name_to_ids.empty()) {
+ for (auto &pair : name_to_ids) {
+ auto it = m_local_to_global_ids.find(pair.second);
+ if (it != m_local_to_global_ids.end()) {
+ // mirrored image must exist within directory to be treated as
+ // a valid image
+ m_image_ids->insert(ImageId(it->second, it->first, pair.first));
+ }
+ }
+
+ if (name_to_ids.size() == MAX_RETURN) {
+ m_start_after = name_to_ids.rbegin()->first;
+ dir_list();
+ return;
+ }
+ }
+
+ finish(0);
+}
+
+template <typename I>
+void RefreshImagesRequest<I>::finish(int r) {
+ dout(10) << "r=" << r << dendl;
+
+ m_on_finish->complete(r);
+ delete this;
+}
+
+} // namespace pool_watcher
+} // namespace mirror
+} // namespace rbd
+
+template class rbd::mirror::pool_watcher::RefreshImagesRequest<librbd::ImageCtx>;
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_RBD_MIRROR_POOL_WATCHER_REFRESH_IMAGES_REQUEST_H
+#define CEPH_RBD_MIRROR_POOL_WATCHER_REFRESH_IMAGES_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/rados/librados.hpp"
+#include "tools/rbd_mirror/types.h"
+#include <string>
+#include <unordered_map>
+
+struct Context;
+
+namespace librbd { struct ImageCtx; }
+
+namespace rbd {
+namespace mirror {
+namespace pool_watcher {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class RefreshImagesRequest {
+public:
+ RefreshImagesRequest *create(librados::IoCtx &remote_io_ctx,
+ ImageIds *image_ids, Context *on_finish) {
+ return new RefreshImagesRequest(remote_io_ctx, image_ids, on_finish);
+ }
+
+ RefreshImagesRequest(librados::IoCtx &remote_io_ctx, ImageIds *image_ids,
+ Context *on_finish)
+ : m_remote_io_ctx(remote_io_ctx), m_image_ids(image_ids),
+ m_on_finish(on_finish) {
+ }
+
+ void send();
+
+private:
+ /**
+ * @verbatim
+ *
+ * <start>
+ * |
+ * | /-------------\
+ * | | |
+ * v v | (more images)
+ * MIRROR_IMAGE_LIST ---/
+ * |
+ * | /-------------\
+ * | | |
+ * v v | (more images)
+ * DIR_LIST ------------/
+ * |
+ * v
+ * <finish>
+ *
+ * @endverbatim
+ */
+
+ typedef std::unordered_map<std::string, std::string> LocalToGlobalIds;
+
+ librados::IoCtx &m_remote_io_ctx;
+ ImageIds *m_image_ids;
+ Context *m_on_finish;
+
+ bufferlist m_out_bl;
+ std::string m_start_after;
+ LocalToGlobalIds m_local_to_global_ids;
+
+ void mirror_image_list();
+ void handle_mirror_image_list(int r);
+
+ void dir_list();
+ void handle_dir_list(int r);
+
+ void finish(int r);
+
+};
+
+} // namespace pool_watcher
+} // namespace mirror
+} // namespace rbd
+
+extern template class rbd::mirror::pool_watcher::RefreshImagesRequest<librbd::ImageCtx>;
+
+#endif // CEPH_RBD_MIRROR_POOL_WATCHER_REFRESH_IMAGES_REQUEST_H
#include <iostream>
#include <memory>
+#include <set>
#include <string>
#include <vector>
+#include <boost/optional.hpp>
#include "include/rbd/librbd.hpp"
#include "ImageSyncThrottler.h"
template <typename I = librbd::ImageCtx>
using ImageSyncThrottlerRef = std::shared_ptr<ImageSyncThrottler<I>>;
+struct ImageId {
+ std::string global_id;
+ std::string id;
+ boost::optional<std::string> name;
+
+ explicit ImageId(const std::string &global_id) : global_id(global_id) {
+ }
+ ImageId(const std::string &global_id, const std::string &id,
+ const boost::optional<std::string> &name = boost::none)
+ : global_id(global_id), id(id), name(name) {
+ }
+
+ inline bool operator==(const ImageId &rhs) const {
+ return (global_id == rhs.global_id && id == rhs.id && name == rhs.name);
+ }
+ inline bool operator<(const ImageId &rhs) const {
+ return global_id < rhs.global_id;
+ }
+};
+
+typedef std::set<ImageId> ImageIds;
+
struct peer_t {
peer_t() = default;
peer_t(const std::string &uuid, const std::string &cluster_name,
projects / ceph.git / commitdiff