ceph/src/test/rbd_mirror/test_mock_InstanceReplayer.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "test/librbd/mock/MockImageCtx.h"
   5 #include "test/rbd_mirror/test_mock_fixture.h"
   6 #include "test/rbd_mirror/mock/MockContextWQ.h"
   7 #include "test/rbd_mirror/mock/MockSafeTimer.h"
   8 #include "tools/rbd_mirror/ImageReplayer.h"
   9 #include "tools/rbd_mirror/InstanceWatcher.h"
  10 #include "tools/rbd_mirror/InstanceReplayer.h"
  11 #include "tools/rbd_mirror/ServiceDaemon.h"
  12 #include "tools/rbd_mirror/Threads.h"
  13 #include "tools/rbd_mirror/image_replayer/Types.h"
  14
  15 namespace librbd {
  16
  17 namespace {
  18
  19 struct MockTestImageCtx : public MockImageCtx {
  20   MockTestImageCtx(librbd::ImageCtx &image_ctx)
  21     : librbd::MockImageCtx(image_ctx) {
  22   }
  23 };
  24
  25 } // anonymous namespace
  26
  27 } // namespace librbd
  28
  29 namespace rbd {
  30 namespace mirror {
  31
  32 template <>
  33 struct Threads<librbd::MockTestImageCtx> {
  34   MockSafeTimer *timer;
  35   ceph::mutex &timer_lock;
  36   ceph::condition_variable timer_cond;
  37
  38   MockContextWQ *work_queue;
  39
  40   Threads(Threads<librbd::ImageCtx> *threads)
  41     : timer(new MockSafeTimer()),
  42       timer_lock(threads->timer_lock),
  43       work_queue(new MockContextWQ()) {
  44   }
  45   ~Threads() {
  46     delete timer;
  47     delete work_queue;
  48   }
  49 };
  50
  51 template<>
  52 struct ServiceDaemon<librbd::MockTestImageCtx> {
  53   MOCK_METHOD4(add_or_update_namespace_attribute,
  54                void(int64_t, const std::string&, const std::string&,
  55                     const service_daemon::AttributeValue&));
  56 };
  57
  58 template<>
  59 struct InstanceWatcher<librbd::MockTestImageCtx> {
  60 };
  61
  62 template<>
  63 struct ImageReplayer<librbd::MockTestImageCtx> {
  64   static ImageReplayer* s_instance;
  65   std::string global_image_id;
  66
  67   static ImageReplayer *create(
  68       librados::IoCtx &local_io_ctx, const std::string &local_mirror_uuid,
  69       const std::string &global_image_id,
  70       Threads<librbd::MockTestImageCtx> *threads,
  71       InstanceWatcher<librbd::MockTestImageCtx> *instance_watcher,
  72       MirrorStatusUpdater<librbd::MockTestImageCtx>* local_status_updater,
  73       journal::CacheManagerHandler *cache_manager_handler,
  74       PoolMetaCache* pool_meta_cache) {
  75     ceph_assert(s_instance != nullptr);
  76     s_instance->global_image_id = global_image_id;
  77     return s_instance;
  78   }
  79
  80   ImageReplayer() {
  81     ceph_assert(s_instance == nullptr);
  82     s_instance = this;
  83   }
  84
  85   virtual ~ImageReplayer() {
  86     ceph_assert(s_instance == this);
  87     s_instance = nullptr;
  88   }
  89
  90   MOCK_METHOD0(destroy, void());
  91   MOCK_METHOD2(start, void(Context *, bool));
  92   MOCK_METHOD2(stop, void(Context *, bool));
  93   MOCK_METHOD1(restart, void(Context*));
  94   MOCK_METHOD0(flush, void());
  95   MOCK_METHOD1(print_status, void(Formatter *));
  96   MOCK_METHOD1(add_peer, void(const Peer<librbd::MockTestImageCtx>& peer));
  97   MOCK_METHOD0(get_global_image_id, const std::string &());
  98   MOCK_METHOD0(get_local_image_id, const std::string &());
  99   MOCK_METHOD0(is_running, bool());
 100   MOCK_METHOD0(is_stopped, bool());
 101   MOCK_METHOD0(is_blocklisted, bool());
 102
 103   MOCK_CONST_METHOD0(is_finished, bool());
 104   MOCK_METHOD1(set_finished, void(bool));
 105
 106   MOCK_CONST_METHOD0(get_health_state, image_replayer::HealthState());
 107 };
 108
 109 ImageReplayer<librbd::MockTestImageCtx>* ImageReplayer<librbd::MockTestImageCtx>::s_instance = nullptr;
 110
 111 template<>
 112 struct MirrorStatusUpdater<librbd::MockTestImageCtx> {
 113 };
 114
 115 } // namespace mirror
 116 } // namespace rbd
 117
 118 // template definitions
 119 #include "tools/rbd_mirror/InstanceReplayer.cc"
 120
 121 namespace rbd {
 122 namespace mirror {
 123
 124 using ::testing::_;
 125 using ::testing::DoAll;
 126 using ::testing::InSequence;
 127 using ::testing::Invoke;
 128 using ::testing::Return;
 129 using ::testing::ReturnArg;
 130 using ::testing::ReturnRef;
 131 using ::testing::WithArg;
 132
 133 class TestMockInstanceReplayer : public TestMockFixture {
 134 public:
 135   typedef Threads<librbd::MockTestImageCtx> MockThreads;
 136   typedef ImageReplayer<librbd::MockTestImageCtx> MockImageReplayer;
 137   typedef InstanceReplayer<librbd::MockTestImageCtx> MockInstanceReplayer;
 138   typedef InstanceWatcher<librbd::MockTestImageCtx> MockInstanceWatcher;
 139   typedef MirrorStatusUpdater<librbd::MockTestImageCtx> MockMirrorStatusUpdater;
 140   typedef ServiceDaemon<librbd::MockTestImageCtx> MockServiceDaemon;
 141
 142   void expect_work_queue(MockThreads &mock_threads) {
 143     EXPECT_CALL(*mock_threads.work_queue, queue(_, _))
 144       .WillOnce(Invoke([this](Context *ctx, int r) {
 145           m_threads->work_queue->queue(ctx, r);
 146         }));
 147   }
 148
 149   void expect_add_event_after(MockThreads &mock_threads,
 150                               Context** timer_ctx = nullptr) {
 151     EXPECT_CALL(*mock_threads.timer, add_event_after(_, _))
 152       .WillOnce(DoAll(
 153         WithArg<1>(Invoke([this, &mock_threads, timer_ctx](Context *ctx) {
 154           ceph_assert(ceph_mutex_is_locked(mock_threads.timer_lock));
 155           if (timer_ctx != nullptr) {
 156             *timer_ctx = ctx;
 157             mock_threads.timer_cond.notify_one();
 158           } else {
 159             m_threads->work_queue->queue(
 160               new LambdaContext([&mock_threads, ctx](int) {
 161                 std::lock_guard timer_lock{mock_threads.timer_lock};
 162                 ctx->complete(0);
 163               }), 0);
 164           }
 165         })),
 166         ReturnArg<1>()));
 167   }
 168
 169   void expect_cancel_event(MockThreads &mock_threads, bool canceled) {
 170     EXPECT_CALL(*mock_threads.timer, cancel_event(_))
 171       .WillOnce(Return(canceled));
 172   }
 173 };
 174
 175 TEST_F(TestMockInstanceReplayer, AcquireReleaseImage) {
 176   MockThreads mock_threads(m_threads);
 177   MockServiceDaemon mock_service_daemon;
 178   MockMirrorStatusUpdater mock_status_updater;
 179   MockInstanceWatcher mock_instance_watcher;
 180   MockImageReplayer mock_image_replayer;
 181   MockInstanceReplayer instance_replayer(
 182       m_local_io_ctx, "local_mirror_uuid",
 183       &mock_threads, &mock_service_daemon, &mock_status_updater, nullptr,
 184       nullptr);
 185   std::string global_image_id("global_image_id");
 186
 187   EXPECT_CALL(mock_image_replayer, get_global_image_id())
 188     .WillRepeatedly(ReturnRef(global_image_id));
 189
 190   InSequence seq;
 191   expect_work_queue(mock_threads);
 192   Context *timer_ctx = nullptr;
 193   expect_add_event_after(mock_threads, &timer_ctx);
 194   instance_replayer.init();
 195   instance_replayer.add_peer({"peer_uuid", m_remote_io_ctx, {}, nullptr});
 196
 197   // Acquire
 198
 199   C_SaferCond on_acquire;
 200   EXPECT_CALL(mock_image_replayer, add_peer(_));
 201   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 202   EXPECT_CALL(mock_image_replayer, is_blocklisted()).WillOnce(Return(false));
 203   EXPECT_CALL(mock_image_replayer, is_finished()).WillOnce(Return(false));
 204   EXPECT_CALL(mock_image_replayer, start(_, false))
 205     .WillOnce(CompleteContext(0));
 206   expect_work_queue(mock_threads);
 207
 208   instance_replayer.acquire_image(&mock_instance_watcher, global_image_id,
 209                                   &on_acquire);
 210   ASSERT_EQ(0, on_acquire.wait());
 211
 212   // Release
 213
 214   C_SaferCond on_release;
 215
 216   EXPECT_CALL(mock_image_replayer, is_stopped())
 217     .WillOnce(Return(false));
 218   EXPECT_CALL(mock_image_replayer, is_running())
 219     .WillOnce(Return(false));
 220   expect_work_queue(mock_threads);
 221   expect_add_event_after(mock_threads);
 222   expect_work_queue(mock_threads);
 223   EXPECT_CALL(mock_image_replayer, is_stopped())
 224     .WillOnce(Return(false));
 225   EXPECT_CALL(mock_image_replayer, is_running())
 226     .WillOnce(Return(true));
 227   EXPECT_CALL(mock_image_replayer, stop(_, false))
 228     .WillOnce(CompleteContext(0));
 229   expect_work_queue(mock_threads);
 230   EXPECT_CALL(mock_image_replayer, is_stopped())
 231     .WillOnce(Return(true));
 232   expect_work_queue(mock_threads);
 233   EXPECT_CALL(mock_image_replayer, destroy());
 234
 235   instance_replayer.release_image("global_image_id", &on_release);
 236   ASSERT_EQ(0, on_release.wait());
 237
 238   expect_work_queue(mock_threads);
 239   expect_cancel_event(mock_threads, true);
 240   expect_work_queue(mock_threads);
 241   instance_replayer.shut_down();
 242   ASSERT_TRUE(timer_ctx != nullptr);
 243   delete timer_ctx;
 244 }
 245
 246 TEST_F(TestMockInstanceReplayer, RemoveFinishedImage) {
 247   MockThreads mock_threads(m_threads);
 248   MockServiceDaemon mock_service_daemon;
 249   MockMirrorStatusUpdater mock_status_updater;
 250   MockInstanceWatcher mock_instance_watcher;
 251   MockImageReplayer mock_image_replayer;
 252   MockInstanceReplayer instance_replayer(
 253       m_local_io_ctx, "local_mirror_uuid",
 254       &mock_threads, &mock_service_daemon, &mock_status_updater, nullptr,
 255       nullptr);
 256   std::string global_image_id("global_image_id");
 257
 258   EXPECT_CALL(mock_image_replayer, get_global_image_id())
 259     .WillRepeatedly(ReturnRef(global_image_id));
 260
 261   InSequence seq;
 262   expect_work_queue(mock_threads);
 263   Context *timer_ctx1 = nullptr;
 264   expect_add_event_after(mock_threads, &timer_ctx1);
 265   instance_replayer.init();
 266   instance_replayer.add_peer({"peer_uuid", m_remote_io_ctx, {}, nullptr});
 267
 268   // Acquire
 269
 270   C_SaferCond on_acquire;
 271   EXPECT_CALL(mock_image_replayer, add_peer(_));
 272   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 273   EXPECT_CALL(mock_image_replayer, is_blocklisted()).WillOnce(Return(false));
 274   EXPECT_CALL(mock_image_replayer, is_finished()).WillOnce(Return(false));
 275   EXPECT_CALL(mock_image_replayer, start(_, false))
 276     .WillOnce(CompleteContext(0));
 277   expect_work_queue(mock_threads);
 278
 279   instance_replayer.acquire_image(&mock_instance_watcher, global_image_id,
 280                                   &on_acquire);
 281   ASSERT_EQ(0, on_acquire.wait());
 282
 283   // periodic start timer
 284   Context *timer_ctx2 = nullptr;
 285   expect_add_event_after(mock_threads, &timer_ctx2);
 286
 287   Context *start_image_replayers_ctx = nullptr;
 288   EXPECT_CALL(*mock_threads.work_queue, queue(_, 0))
 289     .WillOnce(Invoke([&start_image_replayers_ctx](Context *ctx, int r) {
 290                 start_image_replayers_ctx = ctx;
 291               }));
 292
 293   ASSERT_TRUE(timer_ctx1 != nullptr);
 294   {
 295     std::lock_guard timer_locker{mock_threads.timer_lock};
 296     timer_ctx1->complete(0);
 297   }
 298
 299   // remove finished image replayer
 300   EXPECT_CALL(mock_image_replayer, get_health_state()).WillOnce(
 301     Return(image_replayer::HEALTH_STATE_OK));
 302   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 303   EXPECT_CALL(mock_image_replayer, is_blocklisted()).WillOnce(Return(false));
 304   EXPECT_CALL(mock_image_replayer, is_finished()).WillOnce(Return(true));
 305   EXPECT_CALL(mock_image_replayer, destroy());
 306   EXPECT_CALL(mock_service_daemon,
 307               add_or_update_namespace_attribute(_, _, _, _)).Times(3);
 308
 309   ASSERT_TRUE(start_image_replayers_ctx != nullptr);
 310   start_image_replayers_ctx->complete(0);
 311
 312   // shut down
 313   expect_work_queue(mock_threads);
 314   expect_cancel_event(mock_threads, true);
 315   expect_work_queue(mock_threads);
 316   instance_replayer.shut_down();
 317   ASSERT_TRUE(timer_ctx2 != nullptr);
 318   delete timer_ctx2;
 319 }
 320
 321 TEST_F(TestMockInstanceReplayer, Reacquire) {
 322   MockThreads mock_threads(m_threads);
 323   MockServiceDaemon mock_service_daemon;
 324   MockMirrorStatusUpdater mock_status_updater;
 325   MockInstanceWatcher mock_instance_watcher;
 326   MockImageReplayer mock_image_replayer;
 327   MockInstanceReplayer instance_replayer(
 328       m_local_io_ctx, "local_mirror_uuid",
 329       &mock_threads, &mock_service_daemon, &mock_status_updater, nullptr,
 330       nullptr);
 331   std::string global_image_id("global_image_id");
 332
 333   EXPECT_CALL(mock_image_replayer, get_global_image_id())
 334     .WillRepeatedly(ReturnRef(global_image_id));
 335
 336   InSequence seq;
 337   expect_work_queue(mock_threads);
 338   Context *timer_ctx = nullptr;
 339   expect_add_event_after(mock_threads, &timer_ctx);
 340   instance_replayer.init();
 341   instance_replayer.add_peer({"peer_uuid", m_remote_io_ctx, {}, nullptr});
 342
 343   // Acquire
 344
 345   EXPECT_CALL(mock_image_replayer, add_peer(_));
 346   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 347   EXPECT_CALL(mock_image_replayer, is_blocklisted()).WillOnce(Return(false));
 348   EXPECT_CALL(mock_image_replayer, is_finished()).WillOnce(Return(false));
 349   EXPECT_CALL(mock_image_replayer, start(_, false))
 350     .WillOnce(CompleteContext(0));
 351   expect_work_queue(mock_threads);
 352
 353   C_SaferCond on_acquire1;
 354   instance_replayer.acquire_image(&mock_instance_watcher, global_image_id,
 355                                   &on_acquire1);
 356   ASSERT_EQ(0, on_acquire1.wait());
 357
 358   // Re-acquire
 359   EXPECT_CALL(mock_image_replayer, set_finished(false));
 360   EXPECT_CALL(mock_image_replayer, restart(_))
 361     .WillOnce(CompleteContext(0));
 362   expect_work_queue(mock_threads);
 363
 364   C_SaferCond on_acquire2;
 365   instance_replayer.acquire_image(&mock_instance_watcher, global_image_id,
 366                                   &on_acquire2);
 367   ASSERT_EQ(0, on_acquire2.wait());
 368
 369   expect_work_queue(mock_threads);
 370   expect_cancel_event(mock_threads, true);
 371   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 372   expect_work_queue(mock_threads);
 373   expect_work_queue(mock_threads);
 374   EXPECT_CALL(mock_image_replayer, is_stopped()).WillOnce(Return(true));
 375   EXPECT_CALL(mock_image_replayer, destroy());
 376   instance_replayer.shut_down();
 377   ASSERT_TRUE(timer_ctx != nullptr);
 378   delete timer_ctx;
 379 }
 380
 381 } // namespace mirror
 382 } // namespace rbd