ceph/src/test/rbd_mirror/test_mock_PoolReplayer.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 "librbd/api/Config.h"
   5 #include "librbd/api/Namespace.h"
   6 #include "test/librbd/mock/MockImageCtx.h"
   7 #include "test/librados_test_stub/MockTestMemCluster.h"
   8 #include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
   9 #include "test/librados_test_stub/MockTestMemRadosClient.h"
  10 #include "test/rbd_mirror/test_mock_fixture.h"
  11 #include "test/rbd_mirror/mock/MockContextWQ.h"
  12 #include "test/rbd_mirror/mock/MockSafeTimer.h"
  13 #include "tools/rbd_mirror/Throttler.h"
  14 #include "tools/rbd_mirror/LeaderWatcher.h"
  15 #include "tools/rbd_mirror/NamespaceReplayer.h"
  16 #include "tools/rbd_mirror/PoolMetaCache.h"
  17 #include "tools/rbd_mirror/PoolReplayer.h"
  18 #include "tools/rbd_mirror/RemotePoolPoller.h"
  19 #include "tools/rbd_mirror/ServiceDaemon.h"
  20 #include "tools/rbd_mirror/Threads.h"
  21 #include "common/Formatter.h"
  22
  23 namespace librbd {
  24
  25 namespace {
  26
  27 struct MockTestImageCtx : public MockImageCtx {
  28   MockTestImageCtx(librbd::ImageCtx &image_ctx)
  29     : librbd::MockImageCtx(image_ctx) {
  30   }
  31 };
  32
  33 } // anonymous namespace
  34
  35 namespace api {
  36
  37 template <>
  38 class Config<MockTestImageCtx> {
  39 public:
  40   static void apply_pool_overrides(librados::IoCtx& io_ctx,
  41                                    ConfigProxy* config_proxy) {
  42   }
  43 };
  44
  45 template <>
  46 class Namespace<MockTestImageCtx> {
  47 public:
  48   static Namespace* s_instance;
  49
  50   static int list(librados::IoCtx& io_ctx, std::vector<std::string> *names) {
  51     if (s_instance) {
  52       return s_instance->list(names);
  53     }
  54
  55     return 0;
  56   }
  57
  58   Namespace() {
  59     s_instance = this;
  60   }
  61
  62   void add(const std::string &name) {
  63     std::lock_guard locker{m_lock};
  64
  65     m_names.insert(name);
  66   }
  67
  68   void remove(const std::string &name) {
  69     std::lock_guard locker{m_lock};
  70
  71     m_names.erase(name);
  72   }
  73
  74   void clear() {
  75     std::lock_guard locker{m_lock};
  76
  77     m_names.clear();
  78   }
  79
  80 private:
  81   ceph::mutex m_lock = ceph::make_mutex("Namespace");
  82   std::set<std::string> m_names;
  83
  84   int list(std::vector<std::string> *names) {
  85     std::lock_guard locker{m_lock};
  86
  87     names->clear();
  88     names->insert(names->begin(), m_names.begin(), m_names.end());
  89     return 0;
  90   }
  91 };
  92
  93 Namespace<librbd::MockTestImageCtx>* Namespace<librbd::MockTestImageCtx>::s_instance = nullptr;
  94
  95 } // namespace api
  96
  97 } // namespace librbd
  98
  99 namespace rbd {
 100 namespace mirror {
 101
 102 template <>
 103 struct Throttler<librbd::MockTestImageCtx> {
 104   static Throttler* s_instance;
 105
 106   static Throttler *create(
 107       CephContext *cct,
 108       const std::string &max_concurrent_ops_config_param_name) {
 109     return s_instance;
 110   }
 111
 112   Throttler() {
 113     ceph_assert(s_instance == nullptr);
 114     s_instance = this;
 115   }
 116
 117   virtual ~Throttler() {
 118     ceph_assert(s_instance == this);
 119     s_instance = nullptr;
 120   }
 121
 122   MOCK_METHOD1(print_status, void(Formatter*));
 123 };
 124
 125 Throttler<librbd::MockTestImageCtx>* Throttler<librbd::MockTestImageCtx>::s_instance = nullptr;
 126
 127 template <>
 128 struct NamespaceReplayer<librbd::MockTestImageCtx> {
 129   static std::map<std::string, NamespaceReplayer *> s_instances;
 130
 131   static NamespaceReplayer *create(
 132       const std::string &name,
 133       librados::IoCtx &local_ioctx,
 134       librados::IoCtx &remote_ioctx,
 135       const std::string &local_mirror_uuid,
 136       const std::string& local_mirror_peer_uuid,
 137       const RemotePoolMeta& remote_pool_meta,
 138       Threads<librbd::MockTestImageCtx> *threads,
 139       Throttler<librbd::MockTestImageCtx> *image_sync_throttler,
 140       Throttler<librbd::MockTestImageCtx> *image_deletion_throttler,
 141       ServiceDaemon<librbd::MockTestImageCtx> *service_daemon,
 142       journal::CacheManagerHandler *cache_manager_handler,
 143       PoolMetaCache* pool_meta_cache) {
 144     ceph_assert(s_instances.count(name));
 145     auto namespace_replayer = s_instances[name];
 146     s_instances.erase(name);
 147     return namespace_replayer;
 148   }
 149
 150   MOCK_METHOD0(is_blocklisted, bool());
 151   MOCK_METHOD0(get_instance_id, std::string());
 152
 153   MOCK_METHOD1(init, void(Context*));
 154   MOCK_METHOD1(shut_down, void(Context*));
 155
 156   MOCK_METHOD1(handle_acquire_leader, void(Context *));
 157   MOCK_METHOD1(handle_release_leader, void(Context *));
 158   MOCK_METHOD1(handle_update_leader, void(const std::string &));
 159   MOCK_METHOD1(handle_instances_added, void(const std::vector<std::string> &));
 160   MOCK_METHOD1(handle_instances_removed, void(const std::vector<std::string> &));
 161
 162   MOCK_METHOD1(print_status, void(Formatter*));
 163   MOCK_METHOD0(start, void());
 164   MOCK_METHOD0(stop, void());
 165   MOCK_METHOD0(restart, void());
 166   MOCK_METHOD0(flush, void());
 167
 168   NamespaceReplayer(const std::string &name = "") {
 169     ceph_assert(!s_instances.count(name));
 170     s_instances[name] = this;
 171   }
 172 };
 173
 174 std::map<std::string, NamespaceReplayer<librbd::MockTestImageCtx> *> NamespaceReplayer<librbd::MockTestImageCtx>::s_instances;
 175
 176 template<>
 177 struct LeaderWatcher<librbd::MockTestImageCtx> {
 178   static LeaderWatcher* s_instance;
 179   leader_watcher::Listener* listener = nullptr;
 180
 181   static LeaderWatcher *create(Threads<librbd::MockTestImageCtx> *threads,
 182                                librados::IoCtx &ioctx,
 183                                leader_watcher::Listener* listener) {
 184     ceph_assert(s_instance != nullptr);
 185     s_instance->listener = listener;
 186     return s_instance;
 187   }
 188
 189   MOCK_METHOD0(is_blocklisted, bool());
 190   MOCK_METHOD0(is_leader, bool());
 191   MOCK_METHOD0(release_leader, void());
 192
 193   MOCK_METHOD1(get_leader_instance_id, bool(std::string*));
 194   MOCK_METHOD1(list_instances, void(std::vector<std::string>*));
 195
 196   MOCK_METHOD0(init, int());
 197   MOCK_METHOD0(shut_down, int());
 198
 199   LeaderWatcher() {
 200     s_instance = this;
 201   }
 202
 203 };
 204
 205 LeaderWatcher<librbd::MockTestImageCtx>* LeaderWatcher<librbd::MockTestImageCtx>::s_instance = nullptr;
 206
 207 template<>
 208 struct RemotePoolPoller<librbd::MockTestImageCtx> {
 209   static RemotePoolPoller* s_instance;
 210
 211   remote_pool_poller::Listener* listener = nullptr;
 212
 213   static RemotePoolPoller* create(
 214       Threads<librbd::MockTestImageCtx>* threads,
 215       librados::IoCtx& remote_io_ctx,
 216       const std::string& local_site_name,
 217       const std::string& local_mirror_uuid,
 218       remote_pool_poller::Listener& listener) {
 219     ceph_assert(s_instance != nullptr);
 220     s_instance->listener = &listener;
 221     return s_instance;
 222   }
 223
 224   MOCK_METHOD1(init, void(Context*));
 225   MOCK_METHOD1(shut_down, void(Context*));
 226
 227   RemotePoolPoller() {
 228     s_instance = this;
 229   }
 230 };
 231
 232 RemotePoolPoller<librbd::MockTestImageCtx>* RemotePoolPoller<librbd::MockTestImageCtx>::s_instance = nullptr;
 233
 234 template<>
 235 struct ServiceDaemon<librbd::MockTestImageCtx> {
 236   MOCK_METHOD2(add_namespace, void(int64_t, const std::string &));
 237   MOCK_METHOD2(remove_namespace, void(int64_t, const std::string &));
 238
 239   MOCK_METHOD3(add_or_update_attribute,
 240                void(int64_t, const std::string&,
 241                     const service_daemon::AttributeValue&));
 242   MOCK_METHOD2(remove_attribute,
 243                void(int64_t, const std::string&));
 244
 245   MOCK_METHOD4(add_or_update_callout, uint64_t(int64_t, uint64_t,
 246                                                service_daemon::CalloutLevel,
 247                                                const std::string&));
 248   MOCK_METHOD2(remove_callout, void(int64_t, uint64_t));
 249 };
 250
 251 template <>
 252 struct Threads<librbd::MockTestImageCtx> {
 253   MockSafeTimer *timer;
 254   ceph::mutex &timer_lock;
 255   ceph::condition_variable timer_cond;
 256
 257   MockContextWQ *work_queue;
 258
 259   Threads(Threads<librbd::ImageCtx> *threads)
 260     : timer(new MockSafeTimer()),
 261       timer_lock(threads->timer_lock),
 262       work_queue(new MockContextWQ()) {
 263   }
 264   ~Threads() {
 265     delete timer;
 266     delete work_queue;
 267   }
 268 };
 269
 270 } // namespace mirror
 271 } // namespace rbd
 272
 273 // template definitions
 274 #include "tools/rbd_mirror/PoolReplayer.cc"
 275
 276 namespace rbd {
 277 namespace mirror {
 278
 279 using ::testing::_;
 280 using ::testing::AtLeast;
 281 using ::testing::DoAll;
 282 using ::testing::InSequence;
 283 using ::testing::Invoke;
 284 using ::testing::Return;
 285 using ::testing::StrEq;
 286 using ::testing::WithArg;
 287
 288 class TestMockPoolReplayer : public TestMockFixture {
 289 public:
 290   typedef librbd::api::Namespace<librbd::MockTestImageCtx> MockNamespace;
 291   typedef PoolReplayer<librbd::MockTestImageCtx> MockPoolReplayer;
 292   typedef Throttler<librbd::MockTestImageCtx> MockThrottler;
 293   typedef NamespaceReplayer<librbd::MockTestImageCtx> MockNamespaceReplayer;
 294   typedef RemotePoolPoller<librbd::MockTestImageCtx> MockRemotePoolPoller;
 295   typedef LeaderWatcher<librbd::MockTestImageCtx> MockLeaderWatcher;
 296   typedef ServiceDaemon<librbd::MockTestImageCtx> MockServiceDaemon;
 297   typedef Threads<librbd::MockTestImageCtx> MockThreads;
 298
 299   void expect_work_queue(MockThreads &mock_threads) {
 300     EXPECT_CALL(*mock_threads.work_queue, queue(_, _))
 301       .WillRepeatedly(Invoke([this](Context *ctx, int r) {
 302             m_threads->work_queue->queue(ctx, r);
 303           }));
 304   }
 305
 306   void expect_connect(librados::MockTestMemCluster& mock_cluster,
 307                       librados::MockTestMemRadosClient* mock_rados_client,
 308                       const std::string& cluster_name, CephContext** cct_ref) {
 309     EXPECT_CALL(mock_cluster, create_rados_client(_))
 310       .WillOnce(Invoke([cluster_name, mock_rados_client, cct_ref](CephContext* cct) {
 311                   EXPECT_EQ(cluster_name, cct->_conf->cluster);
 312                   if (cct_ref != nullptr) {
 313                     cct->get();
 314                     *cct_ref = cct;
 315                   }
 316                   return mock_rados_client;
 317                 }));
 318   }
 319
 320   void expect_create_ioctx(librados::MockTestMemRadosClient* mock_rados_client,
 321                            librados::MockTestMemIoCtxImpl* mock_io_ctx_impl) {
 322     EXPECT_CALL(*mock_rados_client, create_ioctx(_, _))
 323       .WillOnce(Invoke([mock_io_ctx_impl](int64_t id, const std::string& name) {
 324                   return mock_io_ctx_impl;
 325                 }));
 326   }
 327
 328   void expect_mirror_uuid_get(librados::MockTestMemIoCtxImpl *io_ctx_impl,
 329                               const std::string &uuid, int r) {
 330     bufferlist out_bl;
 331     encode(uuid, out_bl);
 332
 333     EXPECT_CALL(*io_ctx_impl,
 334                 exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_uuid_get"),
 335                      _, _, _, _))
 336       .WillOnce(DoAll(WithArg<5>(Invoke([out_bl](bufferlist *bl) {
 337                           *bl = out_bl;
 338                         })),
 339                       Return(r)));
 340   }
 341
 342   void expect_mirror_mode_get(librados::MockTestMemIoCtxImpl *io_ctx_impl,
 343                               cls::rbd::MirrorMode mirror_mode, int r) {
 344     bufferlist out_bl;
 345     encode(mirror_mode, out_bl);
 346
 347     EXPECT_CALL(*io_ctx_impl,
 348                 exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_mode_get"),
 349                      _, _, _, _))
 350       .WillOnce(DoAll(WithArg<5>(Invoke([out_bl](bufferlist *bl) {
 351                                           *bl = out_bl;
 352                                         })),
 353           Return(r)));
 354   }
 355
 356   void expect_mirror_mode_get(librados::MockTestMemIoCtxImpl *io_ctx_impl) {
 357     EXPECT_CALL(*io_ctx_impl,
 358                 exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_mode_get"),
 359                      _, _, _, _))
 360       .WillRepeatedly(DoAll(WithArg<5>(Invoke([](bufferlist *bl) {
 361                 encode(cls::rbd::MIRROR_MODE_POOL, *bl);
 362               })),
 363           Return(0)));
 364   }
 365
 366   void expect_leader_watcher_init(MockLeaderWatcher& mock_leader_watcher,
 367                                   int r) {
 368     EXPECT_CALL(mock_leader_watcher, init())
 369       .WillOnce(Return(r));
 370   }
 371
 372   void expect_leader_watcher_shut_down(MockLeaderWatcher& mock_leader_watcher) {
 373     EXPECT_CALL(mock_leader_watcher, shut_down());
 374   }
 375
 376   void expect_leader_watcher_get_leader_instance_id(
 377       MockLeaderWatcher& mock_leader_watcher) {
 378     EXPECT_CALL(mock_leader_watcher, get_leader_instance_id(_))
 379       .WillRepeatedly(Return(true));
 380   }
 381
 382   void expect_leader_watcher_list_instances(
 383       MockLeaderWatcher& mock_leader_watcher) {
 384     EXPECT_CALL(mock_leader_watcher, list_instances(_))
 385       .Times(AtLeast(0));
 386   }
 387
 388   void expect_remote_pool_poller_init(
 389       MockRemotePoolPoller& mock_remote_pool_poller,
 390       const RemotePoolMeta& remote_pool_meta, int r) {
 391     EXPECT_CALL(mock_remote_pool_poller, init(_))
 392       .WillOnce(Invoke(
 393                   [this, &mock_remote_pool_poller, remote_pool_meta, r]
 394                   (Context* ctx) {
 395                     if (r >= 0) {
 396                       mock_remote_pool_poller.listener->handle_updated(
 397                         remote_pool_meta);
 398                     }
 399
 400                     m_threads->work_queue->queue(ctx, r);
 401                 }));
 402   }
 403
 404   void expect_remote_pool_poller_shut_down(
 405       MockRemotePoolPoller& mock_remote_pool_poller, int r) {
 406     EXPECT_CALL(mock_remote_pool_poller, shut_down(_))
 407       .WillOnce(Invoke(
 408                   [this, r](Context* ctx) {
 409                     m_threads->work_queue->queue(ctx, r);
 410                 }));
 411   }
 412
 413   void expect_leader_watcher_is_blocklisted(
 414       MockLeaderWatcher &mock_leader_watcher, bool blocklisted) {
 415     EXPECT_CALL(mock_leader_watcher, is_blocklisted())
 416       .WillRepeatedly(Return(blocklisted));
 417   }
 418
 419   void expect_namespace_replayer_is_blocklisted(
 420       MockNamespaceReplayer &mock_namespace_replayer,
 421       bool blocklisted) {
 422     EXPECT_CALL(mock_namespace_replayer, is_blocklisted())
 423       .WillRepeatedly(Return(blocklisted));
 424   }
 425
 426   void expect_namespace_replayer_get_instance_id(
 427       MockNamespaceReplayer &mock_namespace_replayer,
 428       const std::string &instance_id) {
 429     EXPECT_CALL(mock_namespace_replayer, get_instance_id())
 430       .WillOnce(Return(instance_id));
 431   }
 432
 433   void expect_namespace_replayer_init(
 434       MockNamespaceReplayer &mock_namespace_replayer, int r,
 435       Context *on_init = nullptr) {
 436
 437     EXPECT_CALL(mock_namespace_replayer, init(_))
 438       .WillOnce(Invoke([this, r, on_init](Context* ctx) {
 439                          m_threads->work_queue->queue(ctx, r);
 440                          if (on_init != nullptr) {
 441                            m_threads->work_queue->queue(on_init, r);
 442                          }
 443                        }));
 444   }
 445
 446   void expect_namespace_replayer_shut_down(
 447       MockNamespaceReplayer &mock_namespace_replayer,
 448       Context *on_shut_down = nullptr) {
 449     EXPECT_CALL(mock_namespace_replayer, shut_down(_))
 450       .WillOnce(Invoke([this, on_shut_down](Context* ctx) {
 451                          m_threads->work_queue->queue(ctx);
 452                          if (on_shut_down != nullptr) {
 453                            m_threads->work_queue->queue(on_shut_down);
 454                          }
 455                        }));
 456   }
 457
 458   void expect_namespace_replayer_handle_acquire_leader(
 459       MockNamespaceReplayer &mock_namespace_replayer, int r,
 460       Context *on_acquire = nullptr) {
 461     EXPECT_CALL(mock_namespace_replayer, handle_acquire_leader(_))
 462       .WillOnce(Invoke([this, r, on_acquire](Context* ctx) {
 463                          m_threads->work_queue->queue(ctx, r);
 464                          if (on_acquire != nullptr) {
 465                            m_threads->work_queue->queue(on_acquire, r);
 466                          }
 467                        }));
 468   }
 469
 470   void expect_namespace_replayer_handle_release_leader(
 471       MockNamespaceReplayer &mock_namespace_replayer, int r,
 472       Context *on_release = nullptr) {
 473     EXPECT_CALL(mock_namespace_replayer, handle_release_leader(_))
 474       .WillOnce(Invoke([this, r, on_release](Context* ctx) {
 475                          m_threads->work_queue->queue(ctx, r);
 476                          if (on_release != nullptr) {
 477                            m_threads->work_queue->queue(on_release, r);
 478                          }
 479                        }));
 480   }
 481
 482   void expect_namespace_replayer_handle_update_leader(
 483       MockNamespaceReplayer &mock_namespace_replayer,
 484       const std::string &leader_instance_id,
 485       Context *on_update = nullptr) {
 486     EXPECT_CALL(mock_namespace_replayer,
 487                 handle_update_leader(leader_instance_id))
 488       .WillOnce(Invoke([on_update](const std::string &) {
 489                          if (on_update != nullptr) {
 490                            on_update->complete(0);
 491                          }
 492                        }));
 493   }
 494
 495   void expect_namespace_replayer_handle_instances_added(
 496       MockNamespaceReplayer &mock_namespace_replayer) {
 497     EXPECT_CALL(mock_namespace_replayer, handle_instances_added(_));
 498   }
 499
 500   void expect_namespace_replayer_handle_instances_removed(
 501       MockNamespaceReplayer &mock_namespace_replayer) {
 502     EXPECT_CALL(mock_namespace_replayer, handle_instances_removed(_));
 503   }
 504
 505   void expect_service_daemon_add_namespace(
 506       MockServiceDaemon &mock_service_daemon,
 507       const std::string& namespace_name) {
 508     EXPECT_CALL(mock_service_daemon,
 509                 add_namespace(m_local_io_ctx.get_id(), namespace_name));
 510   }
 511
 512   void expect_service_daemon_remove_namespace(
 513       MockServiceDaemon &mock_service_daemon,
 514       const std::string& namespace_name) {
 515     EXPECT_CALL(mock_service_daemon,
 516                 remove_namespace(m_local_io_ctx.get_id(), namespace_name));
 517   }
 518
 519   void expect_service_daemon_add_or_update_attribute(
 520       MockServiceDaemon &mock_service_daemon, const std::string& key,
 521       const service_daemon::AttributeValue& value) {
 522     EXPECT_CALL(mock_service_daemon, add_or_update_attribute(_, key, value));
 523   }
 524
 525   void expect_service_daemon_remove_attribute(
 526       MockServiceDaemon &mock_service_daemon, const std::string& key) {
 527     EXPECT_CALL(mock_service_daemon, remove_attribute(_, key));
 528   }
 529
 530   void expect_service_daemon_add_or_update_instance_id_attribute(
 531       MockServiceDaemon &mock_service_daemon, const std::string &instance_id) {
 532     expect_service_daemon_add_or_update_attribute(
 533         mock_service_daemon, "instance_id", {instance_id});
 534   }
 535
 536   PoolMetaCache m_pool_meta_cache{g_ceph_context};
 537 };
 538
 539 TEST_F(TestMockPoolReplayer, ConfigKeyOverride) {
 540   PeerSpec peer_spec{"uuid", "cluster name", "client.name"};
 541   peer_spec.mon_host = "123";
 542   peer_spec.key = "234";
 543
 544   auto mock_default_namespace_replayer = new MockNamespaceReplayer();
 545   expect_namespace_replayer_is_blocklisted(*mock_default_namespace_replayer,
 546                                            false);
 547
 548   MockThreads mock_threads(m_threads);
 549   expect_work_queue(mock_threads);
 550
 551   auto mock_leader_watcher = new MockLeaderWatcher();
 552   expect_leader_watcher_get_leader_instance_id(*mock_leader_watcher);
 553   expect_leader_watcher_is_blocklisted(*mock_leader_watcher, false);
 554
 555   InSequence seq;
 556
 557   auto& mock_cluster = get_mock_cluster();
 558   auto mock_local_rados_client = mock_cluster.do_create_rados_client(
 559     g_ceph_context);
 560   expect_connect(mock_cluster, mock_local_rados_client, "ceph", nullptr);
 561
 562   auto mock_remote_rados_client = mock_cluster.do_create_rados_client(
 563     g_ceph_context);
 564   CephContext* remote_cct = nullptr;
 565   expect_connect(mock_cluster, mock_remote_rados_client, "cluster name",
 566                  &remote_cct);
 567
 568   auto mock_local_io_ctx = mock_local_rados_client->do_create_ioctx(
 569     m_local_io_ctx.get_id(), m_local_io_ctx.get_pool_name());
 570   expect_create_ioctx(mock_local_rados_client, mock_local_io_ctx);
 571
 572   expect_mirror_uuid_get(mock_local_io_ctx, "uuid", 0);
 573   auto mock_remote_pool_poller = new MockRemotePoolPoller();
 574   expect_remote_pool_poller_init(*mock_remote_pool_poller,
 575                                  {"remote mirror uuid", ""}, 0);
 576   expect_namespace_replayer_init(*mock_default_namespace_replayer, 0);
 577   expect_leader_watcher_init(*mock_leader_watcher, 0);
 578
 579   MockServiceDaemon mock_service_daemon;
 580   std::string instance_id = stringify(mock_local_io_ctx->get_instance_id());
 581   expect_service_daemon_add_or_update_instance_id_attribute(
 582     mock_service_daemon, instance_id);
 583
 584   MockPoolReplayer pool_replayer(&mock_threads, &mock_service_daemon, nullptr,
 585                                  &m_pool_meta_cache,
 586                                  m_local_io_ctx.get_id(), peer_spec, {});
 587   pool_replayer.init("siteA");
 588
 589   ASSERT_TRUE(remote_cct != nullptr);
 590   ASSERT_EQ("123", remote_cct->_conf.get_val<std::string>("mon_host"));
 591   ASSERT_EQ("234", remote_cct->_conf.get_val<std::string>("key"));
 592   remote_cct->put();
 593
 594   expect_leader_watcher_shut_down(*mock_leader_watcher);
 595   expect_namespace_replayer_shut_down(*mock_default_namespace_replayer);
 596   expect_remote_pool_poller_shut_down(*mock_remote_pool_poller, 0);
 597
 598   pool_replayer.shut_down();
 599 }
 600
 601 TEST_F(TestMockPoolReplayer, AcquireReleaseLeader) {
 602   PeerSpec peer_spec{"uuid", "cluster name", "client.name"};
 603   peer_spec.mon_host = "123";
 604   peer_spec.key = "234";
 605
 606   auto mock_default_namespace_replayer = new MockNamespaceReplayer();
 607   expect_namespace_replayer_is_blocklisted(*mock_default_namespace_replayer,
 608                                            false);
 609
 610   MockThreads mock_threads(m_threads);
 611   expect_work_queue(mock_threads);
 612
 613   auto mock_leader_watcher = new MockLeaderWatcher();
 614   expect_leader_watcher_get_leader_instance_id(*mock_leader_watcher);
 615   expect_leader_watcher_list_instances(*mock_leader_watcher);
 616   expect_leader_watcher_is_blocklisted(*mock_leader_watcher, false);
 617
 618   InSequence seq;
 619
 620   auto& mock_cluster = get_mock_cluster();
 621   auto mock_local_rados_client = mock_cluster.do_create_rados_client(
 622     g_ceph_context);
 623   expect_connect(mock_cluster, mock_local_rados_client, "ceph", nullptr);
 624
 625   auto mock_remote_rados_client = mock_cluster.do_create_rados_client(
 626     g_ceph_context);
 627   expect_connect(mock_cluster, mock_remote_rados_client, "cluster name",
 628                  nullptr);
 629
 630   auto mock_local_io_ctx = mock_local_rados_client->do_create_ioctx(
 631     m_local_io_ctx.get_id(), m_local_io_ctx.get_pool_name());
 632   expect_create_ioctx(mock_local_rados_client, mock_local_io_ctx);
 633
 634   expect_mirror_uuid_get(mock_local_io_ctx, "uuid", 0);
 635   auto mock_remote_pool_poller = new MockRemotePoolPoller();
 636   expect_remote_pool_poller_init(*mock_remote_pool_poller,
 637                                  {"remote mirror uuid", ""}, 0);
 638   expect_namespace_replayer_init(*mock_default_namespace_replayer, 0);
 639   expect_leader_watcher_init(*mock_leader_watcher, 0);
 640
 641   MockServiceDaemon mock_service_daemon;
 642   std::string instance_id = stringify(mock_local_io_ctx->get_instance_id());
 643   expect_service_daemon_add_or_update_instance_id_attribute(
 644     mock_service_daemon, instance_id);
 645
 646   MockPoolReplayer pool_replayer(&mock_threads, &mock_service_daemon, nullptr,
 647                                  &m_pool_meta_cache,
 648                                  m_local_io_ctx.get_id(), peer_spec, {});
 649   pool_replayer.init("siteA");
 650
 651   expect_service_daemon_add_or_update_attribute(
 652       mock_service_daemon, SERVICE_DAEMON_LEADER_KEY, true);
 653   expect_namespace_replayer_handle_acquire_leader(
 654       *mock_default_namespace_replayer, 0);
 655
 656   C_SaferCond on_acquire;
 657   mock_leader_watcher->listener->post_acquire_handler(&on_acquire);
 658   ASSERT_EQ(0, on_acquire.wait());
 659
 660   expect_service_daemon_remove_attribute(mock_service_daemon,
 661                                          SERVICE_DAEMON_LEADER_KEY);
 662   expect_namespace_replayer_handle_release_leader(
 663       *mock_default_namespace_replayer, 0);
 664
 665   C_SaferCond on_release;
 666   mock_leader_watcher->listener->pre_release_handler(&on_release);
 667   ASSERT_EQ(0, on_release.wait());
 668
 669   expect_leader_watcher_shut_down(*mock_leader_watcher);
 670   expect_namespace_replayer_shut_down(*mock_default_namespace_replayer);
 671   expect_remote_pool_poller_shut_down(*mock_remote_pool_poller, 0);
 672
 673   pool_replayer.shut_down();
 674 }
 675
 676 TEST_F(TestMockPoolReplayer, Namespaces) {
 677   PeerSpec peer_spec{"uuid", "cluster name", "client.name"};
 678   peer_spec.mon_host = "123";
 679   peer_spec.key = "234";
 680
 681   g_ceph_context->_conf.set_val(
 682       "rbd_mirror_pool_replayers_refresh_interval", "1");
 683
 684   MockNamespace mock_namespace;
 685
 686   auto mock_default_namespace_replayer = new MockNamespaceReplayer();
 687   expect_namespace_replayer_is_blocklisted(*mock_default_namespace_replayer,
 688                                            false);
 689
 690   auto mock_ns1_namespace_replayer = new MockNamespaceReplayer("ns1");
 691   expect_namespace_replayer_is_blocklisted(*mock_ns1_namespace_replayer,
 692                                            false);
 693
 694   auto mock_ns2_namespace_replayer = new MockNamespaceReplayer("ns2");
 695   expect_namespace_replayer_is_blocklisted(*mock_ns2_namespace_replayer,
 696                                            false);
 697
 698   MockThreads mock_threads(m_threads);
 699   expect_work_queue(mock_threads);
 700
 701   auto mock_leader_watcher = new MockLeaderWatcher();
 702   expect_leader_watcher_get_leader_instance_id(*mock_leader_watcher);
 703   expect_leader_watcher_list_instances(*mock_leader_watcher);
 704   expect_leader_watcher_is_blocklisted(*mock_leader_watcher, false);
 705
 706   auto& mock_cluster = get_mock_cluster();
 707   auto mock_local_rados_client = mock_cluster.do_create_rados_client(
 708       g_ceph_context);
 709   auto mock_local_io_ctx = mock_local_rados_client->do_create_ioctx(
 710       m_local_io_ctx.get_id(), m_local_io_ctx.get_pool_name());
 711   auto mock_remote_rados_client = mock_cluster.do_create_rados_client(
 712       g_ceph_context);
 713
 714   expect_mirror_mode_get(mock_local_io_ctx);
 715
 716   InSequence seq;
 717
 718   expect_connect(mock_cluster, mock_local_rados_client, "ceph", nullptr);
 719   expect_connect(mock_cluster, mock_remote_rados_client, "cluster name",
 720                  nullptr);
 721   expect_create_ioctx(mock_local_rados_client, mock_local_io_ctx);
 722   expect_mirror_uuid_get(mock_local_io_ctx, "uuid", 0);
 723   auto mock_remote_pool_poller = new MockRemotePoolPoller();
 724   expect_remote_pool_poller_init(*mock_remote_pool_poller,
 725                                  {"remote mirror uuid", ""}, 0);
 726   expect_namespace_replayer_init(*mock_default_namespace_replayer, 0);
 727   expect_leader_watcher_init(*mock_leader_watcher, 0);
 728
 729   MockServiceDaemon mock_service_daemon;
 730   std::string instance_id = stringify(mock_local_io_ctx->get_instance_id());
 731   expect_service_daemon_add_or_update_instance_id_attribute(
 732     mock_service_daemon, instance_id);
 733
 734   MockPoolReplayer pool_replayer(&mock_threads, &mock_service_daemon, nullptr,
 735                                  &m_pool_meta_cache,
 736                                  m_local_io_ctx.get_id(), peer_spec, {});
 737   pool_replayer.init("siteA");
 738
 739   C_SaferCond on_ns1_init;
 740   expect_namespace_replayer_init(*mock_ns1_namespace_replayer, 0);
 741   expect_service_daemon_add_namespace(mock_service_daemon, "ns1");
 742   expect_namespace_replayer_handle_update_leader(*mock_ns1_namespace_replayer,
 743                                                  "", &on_ns1_init);
 744
 745   mock_namespace.add("ns1");
 746   ASSERT_EQ(0, on_ns1_init.wait());
 747
 748   expect_service_daemon_add_or_update_attribute(
 749       mock_service_daemon, SERVICE_DAEMON_LEADER_KEY, true);
 750   expect_namespace_replayer_handle_acquire_leader(
 751       *mock_default_namespace_replayer, 0);
 752   expect_namespace_replayer_handle_acquire_leader(
 753       *mock_ns1_namespace_replayer, 0);
 754
 755   C_SaferCond on_acquire;
 756   mock_leader_watcher->listener->post_acquire_handler(&on_acquire);
 757   ASSERT_EQ(0, on_acquire.wait());
 758
 759   expect_namespace_replayer_init(*mock_ns2_namespace_replayer, 0);
 760   expect_service_daemon_add_namespace(mock_service_daemon, "ns2");
 761   C_SaferCond on_ns2_acquire;
 762   expect_namespace_replayer_handle_acquire_leader(
 763       *mock_ns2_namespace_replayer, 0, &on_ns2_acquire);
 764   expect_namespace_replayer_handle_instances_added(
 765       *mock_ns2_namespace_replayer);
 766
 767   mock_namespace.add("ns2");
 768   ASSERT_EQ(0, on_ns2_acquire.wait());
 769
 770   C_SaferCond on_ns2_shut_down;
 771   expect_service_daemon_remove_namespace(mock_service_daemon, "ns2");
 772   expect_namespace_replayer_shut_down(*mock_ns2_namespace_replayer,
 773                                       &on_ns2_shut_down);
 774   mock_namespace.remove("ns2");
 775   ASSERT_EQ(0, on_ns2_shut_down.wait());
 776
 777   expect_service_daemon_remove_attribute(mock_service_daemon,
 778                                          SERVICE_DAEMON_LEADER_KEY);
 779   expect_namespace_replayer_handle_release_leader(
 780       *mock_default_namespace_replayer, 0);
 781   expect_namespace_replayer_handle_release_leader(
 782       *mock_ns1_namespace_replayer, 0);
 783
 784   C_SaferCond on_release;
 785   mock_leader_watcher->listener->pre_release_handler(&on_release);
 786   ASSERT_EQ(0, on_release.wait());
 787
 788   expect_service_daemon_remove_namespace(mock_service_daemon, "ns1");
 789   expect_namespace_replayer_shut_down(*mock_ns1_namespace_replayer);
 790   expect_leader_watcher_shut_down(*mock_leader_watcher);
 791   expect_namespace_replayer_shut_down(*mock_default_namespace_replayer);
 792   expect_remote_pool_poller_shut_down(*mock_remote_pool_poller, 0);
 793
 794   pool_replayer.shut_down();
 795 }
 796
 797 TEST_F(TestMockPoolReplayer, NamespacesError) {
 798   PeerSpec peer_spec{"uuid", "cluster name", "client.name"};
 799   peer_spec.mon_host = "123";
 800   peer_spec.key = "234";
 801
 802   g_ceph_context->_conf.set_val(
 803       "rbd_mirror_pool_replayers_refresh_interval", "1");
 804
 805   MockNamespace mock_namespace;
 806
 807   auto mock_default_namespace_replayer = new MockNamespaceReplayer();
 808   expect_namespace_replayer_is_blocklisted(*mock_default_namespace_replayer,
 809                                            false);
 810   auto mock_ns1_namespace_replayer = new MockNamespaceReplayer("ns1");
 811   auto mock_ns2_namespace_replayer = new MockNamespaceReplayer("ns2");
 812   expect_namespace_replayer_is_blocklisted(*mock_ns2_namespace_replayer,
 813                                            false);
 814   auto mock_ns3_namespace_replayer = new MockNamespaceReplayer("ns3");
 815
 816   MockThreads mock_threads(m_threads);
 817   expect_work_queue(mock_threads);
 818
 819   auto mock_leader_watcher = new MockLeaderWatcher();
 820   expect_leader_watcher_get_leader_instance_id(*mock_leader_watcher);
 821   expect_leader_watcher_list_instances(*mock_leader_watcher);
 822   expect_leader_watcher_is_blocklisted(*mock_leader_watcher, false);
 823
 824   auto& mock_cluster = get_mock_cluster();
 825   auto mock_local_rados_client = mock_cluster.do_create_rados_client(
 826       g_ceph_context);
 827   auto mock_local_io_ctx = mock_local_rados_client->do_create_ioctx(
 828       m_local_io_ctx.get_id(), m_local_io_ctx.get_pool_name());
 829   auto mock_remote_rados_client = mock_cluster.do_create_rados_client(
 830       g_ceph_context);
 831
 832   expect_mirror_mode_get(mock_local_io_ctx);
 833
 834   InSequence seq;
 835
 836   expect_connect(mock_cluster, mock_local_rados_client, "ceph", nullptr);
 837   expect_connect(mock_cluster, mock_remote_rados_client, "cluster name",
 838                  nullptr);
 839   expect_create_ioctx(mock_local_rados_client, mock_local_io_ctx);
 840   expect_mirror_uuid_get(mock_local_io_ctx, "uuid", 0);
 841   auto mock_remote_pool_poller = new MockRemotePoolPoller();
 842   expect_remote_pool_poller_init(*mock_remote_pool_poller,
 843                                  {"remote mirror uuid", ""}, 0);
 844   expect_namespace_replayer_init(*mock_default_namespace_replayer, 0);
 845   expect_leader_watcher_init(*mock_leader_watcher, 0);
 846
 847   MockServiceDaemon mock_service_daemon;
 848   std::string instance_id = stringify(mock_local_io_ctx->get_instance_id());
 849   expect_service_daemon_add_or_update_instance_id_attribute(
 850     mock_service_daemon, instance_id);
 851
 852   MockPoolReplayer pool_replayer(&mock_threads, &mock_service_daemon, nullptr,
 853                                  &m_pool_meta_cache,
 854                                  m_local_io_ctx.get_id(), peer_spec, {});
 855   pool_replayer.init("siteA");
 856
 857   // test namespace replayer init fails for non leader
 858
 859   C_SaferCond on_ns1_init;
 860   Context* ctx = new LambdaContext(
 861       [&mock_namespace, &on_ns1_init](int r) {
 862         mock_namespace.remove("ns1");
 863         on_ns1_init.complete(r);
 864       });
 865   expect_namespace_replayer_init(*mock_ns1_namespace_replayer, -EINVAL, ctx);
 866   mock_namespace.add("ns1");
 867   ASSERT_EQ(-EINVAL, on_ns1_init.wait());
 868
 869   // test acquire leader fails when default namespace replayer fails
 870
 871   expect_service_daemon_add_or_update_attribute(
 872     mock_service_daemon, SERVICE_DAEMON_LEADER_KEY, true);
 873   expect_namespace_replayer_handle_acquire_leader(
 874     *mock_default_namespace_replayer, -EINVAL);
 875
 876   C_SaferCond on_acquire1;
 877   mock_leader_watcher->listener->post_acquire_handler(&on_acquire1);
 878   ASSERT_EQ(-EINVAL, on_acquire1.wait());
 879
 880   // test acquire leader succeeds when non-default namespace replayer fails
 881
 882   C_SaferCond on_ns2_init;
 883   expect_namespace_replayer_init(*mock_ns2_namespace_replayer, 0);
 884   expect_service_daemon_add_namespace(mock_service_daemon, "ns2");
 885   expect_namespace_replayer_handle_update_leader(*mock_ns2_namespace_replayer,
 886                                                  "", &on_ns2_init);
 887   mock_namespace.add("ns2");
 888   ASSERT_EQ(0, on_ns2_init.wait());
 889
 890   expect_service_daemon_add_or_update_attribute(
 891       mock_service_daemon, SERVICE_DAEMON_LEADER_KEY, true);
 892   expect_namespace_replayer_handle_acquire_leader(
 893       *mock_default_namespace_replayer, 0);
 894
 895   expect_namespace_replayer_handle_acquire_leader(*mock_ns2_namespace_replayer,
 896                                                   -EINVAL);
 897   ctx = new LambdaContext(
 898       [&mock_namespace](int) {
 899         mock_namespace.remove("ns2");
 900       });
 901   expect_service_daemon_remove_namespace(mock_service_daemon, "ns2");
 902   expect_namespace_replayer_shut_down(*mock_ns2_namespace_replayer, ctx);
 903   mock_namespace.add("ns2");
 904
 905   C_SaferCond on_acquire2;
 906   mock_leader_watcher->listener->post_acquire_handler(&on_acquire2);
 907   ASSERT_EQ(0, on_acquire2.wait());
 908
 909   // test namespace replayer init fails on acquire leader
 910
 911   C_SaferCond on_ns3_shut_down;
 912   ctx = new LambdaContext(
 913       [&mock_namespace, &on_ns3_shut_down](int) {
 914         mock_namespace.remove("ns3");
 915         on_ns3_shut_down.complete(0);
 916       });
 917   expect_namespace_replayer_init(*mock_ns3_namespace_replayer, 0);
 918   expect_service_daemon_add_namespace(mock_service_daemon, "ns3");
 919   expect_namespace_replayer_handle_acquire_leader(*mock_ns3_namespace_replayer,
 920                                                   -EINVAL);
 921   expect_service_daemon_remove_namespace(mock_service_daemon, "ns3");
 922   expect_namespace_replayer_shut_down(*mock_ns3_namespace_replayer, ctx);
 923   mock_namespace.add("ns3");
 924   ASSERT_EQ(0, on_ns3_shut_down.wait());
 925
 926   expect_leader_watcher_shut_down(*mock_leader_watcher);
 927   expect_namespace_replayer_shut_down(*mock_default_namespace_replayer);
 928   expect_remote_pool_poller_shut_down(*mock_remote_pool_poller, 0);
 929
 930   pool_replayer.shut_down();
 931 }
 932
 933 } // namespace mirror
 934 } // namespace rbd