src/ceph/src/test/rbd_mirror/test_ImageReplayer.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph distributed storage system
   5  *
   6  * Copyright (C) 2016 Mirantis Inc
   7  *
   8  * Author: Mykola Golub <mgolub@mirantis.com>
   9  *
  10  *  This library is free software; you can redistribute it and/or
  11  *  modify it under the terms of the GNU Lesser General Public
  12  *  License as published by the Free Software Foundation; either
  13  *  version 2.1 of the License, or (at your option) any later version.
  14  *
  15  */
  16
  17 #include "include/rados/librados.hpp"
  18 #include "include/rbd/librbd.hpp"
  19 #include "include/stringify.h"
  20 #include "test/rbd_mirror/test_fixture.h"
  21 #include "cls/journal/cls_journal_types.h"
  22 #include "cls/journal/cls_journal_client.h"
  23 #include "cls/rbd/cls_rbd_types.h"
  24 #include "cls/rbd/cls_rbd_client.h"
  25 #include "journal/Journaler.h"
  26 #include "librbd/ExclusiveLock.h"
  27 #include "librbd/ImageCtx.h"
  28 #include "librbd/ImageState.h"
  29 #include "librbd/Journal.h"
  30 #include "librbd/Operations.h"
  31 #include "librbd/Utils.h"
  32 #include "librbd/internal.h"
  33 #include "librbd/api/Mirror.h"
  34 #include "librbd/io/AioCompletion.h"
  35 #include "librbd/io/ImageRequestWQ.h"
  36 #include "librbd/io/ReadResult.h"
  37 #include "tools/rbd_mirror/types.h"
  38 #include "tools/rbd_mirror/ImageDeleter.h"
  39 #include "tools/rbd_mirror/ImageReplayer.h"
  40 #include "tools/rbd_mirror/InstanceWatcher.h"
  41 #include "tools/rbd_mirror/ServiceDaemon.h"
  42 #include "tools/rbd_mirror/Threads.h"
  43
  44 #include "test/librados/test.h"
  45 #include "gtest/gtest.h"
  46
  47 using rbd::mirror::RadosRef;
  48
  49 void register_test_rbd_mirror() {
  50 }
  51
  52 #define TEST_IO_SIZE 512
  53 #define TEST_IO_COUNT 11
  54
  55 class TestImageReplayer : public ::rbd::mirror::TestFixture {
  56 public:
  57   struct C_WatchCtx : public librados::WatchCtx2 {
  58     TestImageReplayer *test;
  59     std::string oid;
  60     Mutex lock;
  61     Cond cond;
  62     bool notified;
  63
  64     C_WatchCtx(TestImageReplayer *test, const std::string &oid)
  65       : test(test), oid(oid), lock("C_WatchCtx::lock"), notified(false) {
  66     }
  67
  68     void handle_notify(uint64_t notify_id, uint64_t cookie,
  69                                uint64_t notifier_id, bufferlist& bl_) override {
  70       bufferlist bl;
  71       test->m_remote_ioctx.notify_ack(oid, notify_id, cookie, bl);
  72
  73       Mutex::Locker locker(lock);
  74       notified = true;
  75       cond.Signal();
  76     }
  77
  78     void handle_error(uint64_t cookie, int err) override {
  79       ASSERT_EQ(0, err);
  80     }
  81   };
  82
  83   TestImageReplayer()
  84     : m_local_cluster(new librados::Rados()), m_watch_handle(0)
  85   {
  86     EXPECT_EQ("", connect_cluster_pp(*m_local_cluster.get()));
  87     EXPECT_EQ(0, m_local_cluster->conf_set("rbd_cache", "false"));
  88     EXPECT_EQ(0, m_local_cluster->conf_set("rbd_mirror_journal_poll_age", "1"));
  89
  90     m_local_pool_name = get_temp_pool_name();
  91     EXPECT_EQ(0, m_local_cluster->pool_create(m_local_pool_name.c_str()));
  92     EXPECT_EQ(0, m_local_cluster->ioctx_create(m_local_pool_name.c_str(),
  93                                               m_local_ioctx));
  94     m_local_ioctx.application_enable("rbd", true);
  95
  96     EXPECT_EQ("", connect_cluster_pp(m_remote_cluster));
  97     EXPECT_EQ(0, m_remote_cluster.conf_set("rbd_cache", "false"));
  98
  99     m_remote_pool_name = get_temp_pool_name();
 100     EXPECT_EQ(0, m_remote_cluster.pool_create(m_remote_pool_name.c_str()));
 101     m_remote_pool_id = m_remote_cluster.pool_lookup(m_remote_pool_name.c_str());
 102     EXPECT_GE(m_remote_pool_id, 0);
 103
 104     EXPECT_EQ(0, m_remote_cluster.ioctx_create(m_remote_pool_name.c_str(),
 105                                                m_remote_ioctx));
 106     m_remote_ioctx.application_enable("rbd", true);
 107
 108     EXPECT_EQ(0, librbd::api::Mirror<>::mode_set(m_remote_ioctx,
 109                                                  RBD_MIRROR_MODE_POOL));
 110
 111     m_image_name = get_temp_image_name();
 112     uint64_t features = librbd::util::get_rbd_default_features(g_ceph_context);
 113     features |= RBD_FEATURE_EXCLUSIVE_LOCK | RBD_FEATURE_JOURNALING;
 114     int order = 0;
 115     EXPECT_EQ(0, librbd::create(m_remote_ioctx, m_image_name.c_str(), 1 << 22,
 116                                 false, features, &order, 0, 0));
 117     m_remote_image_id = get_image_id(m_remote_ioctx, m_image_name);
 118     m_global_image_id = get_global_image_id(m_remote_ioctx, m_remote_image_id);
 119
 120     m_threads.reset(new rbd::mirror::Threads<>(reinterpret_cast<CephContext*>(
 121       m_local_ioctx.cct())));
 122
 123     m_service_daemon.reset(new rbd::mirror::ServiceDaemon<>(g_ceph_context,
 124                                                             m_local_cluster,
 125                                                             m_threads.get()));
 126     m_image_deleter.reset(new rbd::mirror::ImageDeleter<>(
 127       m_threads->work_queue, m_threads->timer, &m_threads->timer_lock,
 128       m_service_daemon.get()));
 129     m_instance_watcher = rbd::mirror::InstanceWatcher<>::create(
 130         m_local_ioctx, m_threads->work_queue, nullptr);
 131     m_instance_watcher->handle_acquire_leader();
 132   }
 133
 134   ~TestImageReplayer() override
 135   {
 136     unwatch();
 137
 138     m_instance_watcher->handle_release_leader();
 139
 140     delete m_replayer;
 141     delete m_instance_watcher;
 142
 143     EXPECT_EQ(0, m_remote_cluster.pool_delete(m_remote_pool_name.c_str()));
 144     EXPECT_EQ(0, m_local_cluster->pool_delete(m_local_pool_name.c_str()));
 145   }
 146
 147   template <typename ImageReplayerT = rbd::mirror::ImageReplayer<> >
 148   void create_replayer() {
 149     m_replayer = new ImageReplayerT(
 150         m_threads.get(), m_image_deleter.get(), m_instance_watcher,
 151         rbd::mirror::RadosRef(new librados::Rados(m_local_ioctx)),
 152         m_local_mirror_uuid, m_local_ioctx.get_id(), m_global_image_id);
 153     m_replayer->add_peer("peer uuid", m_remote_ioctx);
 154   }
 155
 156   void start()
 157   {
 158     C_SaferCond cond;
 159     m_replayer->start(&cond);
 160     ASSERT_EQ(0, cond.wait());
 161
 162     ASSERT_EQ(0U, m_watch_handle);
 163     std::string oid = ::journal::Journaler::header_oid(m_remote_image_id);
 164     m_watch_ctx = new C_WatchCtx(this, oid);
 165     ASSERT_EQ(0, m_remote_ioctx.watch2(oid, &m_watch_handle, m_watch_ctx));
 166   }
 167
 168   void unwatch() {
 169     if (m_watch_handle != 0) {
 170       m_remote_ioctx.unwatch2(m_watch_handle);
 171       delete m_watch_ctx;
 172       m_watch_ctx = nullptr;
 173       m_watch_handle = 0;
 174     }
 175   }
 176
 177   void stop()
 178   {
 179     unwatch();
 180
 181     C_SaferCond cond;
 182     m_replayer->stop(&cond);
 183     ASSERT_EQ(0, cond.wait());
 184   }
 185
 186   void bootstrap()
 187   {
 188     create_replayer<>();
 189
 190     start();
 191     wait_for_replay_complete();
 192     stop();
 193   }
 194
 195   std::string  get_temp_image_name()
 196   {
 197     return "image" + stringify(++_image_number);
 198   }
 199
 200   std::string get_image_id(librados::IoCtx &ioctx, const string &image_name)
 201   {
 202     std::string obj = librbd::util::id_obj_name(image_name);
 203     std::string id;
 204     EXPECT_EQ(0, librbd::cls_client::get_id(&ioctx, obj, &id));
 205     return id;
 206   }
 207
 208   std::string get_global_image_id(librados::IoCtx& io_ctx,
 209                                   const std::string& image_id) {
 210     cls::rbd::MirrorImage mirror_image;
 211     EXPECT_EQ(0, librbd::cls_client::mirror_image_get(&io_ctx, image_id,
 212                                                       &mirror_image));
 213     return mirror_image.global_image_id;
 214   }
 215
 216   void open_image(librados::IoCtx &ioctx, const std::string &image_name,
 217                   bool readonly, librbd::ImageCtx **ictxp)
 218   {
 219     librbd::ImageCtx *ictx = new librbd::ImageCtx(image_name.c_str(),
 220                                                   "", "", ioctx, readonly);
 221     EXPECT_EQ(0, ictx->state->open(false));
 222     *ictxp = ictx;
 223   }
 224
 225   void open_local_image(librbd::ImageCtx **ictxp)
 226   {
 227     open_image(m_local_ioctx, m_image_name, true, ictxp);
 228   }
 229
 230   void open_remote_image(librbd::ImageCtx **ictxp)
 231   {
 232     open_image(m_remote_ioctx, m_image_name, false, ictxp);
 233   }
 234
 235   void close_image(librbd::ImageCtx *ictx)
 236   {
 237     ictx->state->close();
 238   }
 239
 240   void get_commit_positions(cls::journal::ObjectPosition *master_position,
 241                             cls::journal::ObjectPosition *mirror_position)
 242   {
 243     std::string master_client_id = "";
 244     std::string mirror_client_id = m_local_mirror_uuid;
 245
 246     C_SaferCond cond;
 247     uint64_t minimum_set;
 248     uint64_t active_set;
 249     std::set<cls::journal::Client> registered_clients;
 250     std::string oid = ::journal::Journaler::header_oid(m_remote_image_id);
 251     cls::journal::client::get_mutable_metadata(m_remote_ioctx, oid,
 252                                                &minimum_set, &active_set,
 253                                                &registered_clients, &cond);
 254     ASSERT_EQ(0, cond.wait());
 255
 256     *master_position = cls::journal::ObjectPosition();
 257     *mirror_position = cls::journal::ObjectPosition();
 258
 259     std::set<cls::journal::Client>::const_iterator c;
 260     for (c = registered_clients.begin(); c != registered_clients.end(); ++c) {
 261       std::cout << __func__ << ": client: " << *c << std::endl;
 262       if (c->state != cls::journal::CLIENT_STATE_CONNECTED) {
 263         continue;
 264       }
 265       cls::journal::ObjectPositions object_positions =
 266         c->commit_position.object_positions;
 267       cls::journal::ObjectPositions::const_iterator p =
 268         object_positions.begin();
 269       if (p != object_positions.end()) {
 270         if (c->id == master_client_id) {
 271           ASSERT_EQ(cls::journal::ObjectPosition(), *master_position);
 272           *master_position = *p;
 273         } else if (c->id == mirror_client_id) {
 274           ASSERT_EQ(cls::journal::ObjectPosition(), *mirror_position);
 275           *mirror_position = *p;
 276         }
 277       }
 278     }
 279   }
 280
 281   bool wait_for_watcher_notify(int seconds)
 282   {
 283     if (m_watch_handle == 0) {
 284       return false;
 285     }
 286
 287     Mutex::Locker locker(m_watch_ctx->lock);
 288     while (!m_watch_ctx->notified) {
 289       if (m_watch_ctx->cond.WaitInterval(m_watch_ctx->lock,
 290                                          utime_t(seconds, 0)) != 0) {
 291         return false;
 292       }
 293     }
 294     m_watch_ctx->notified = false;
 295     return true;
 296   }
 297
 298   void wait_for_replay_complete()
 299   {
 300     cls::journal::ObjectPosition master_position;
 301     cls::journal::ObjectPosition mirror_position;
 302
 303     for (int i = 0; i < 100; i++) {
 304       printf("m_replayer->flush()\n");
 305       C_SaferCond cond;
 306       m_replayer->flush(&cond);
 307       ASSERT_EQ(0, cond.wait());
 308       get_commit_positions(&master_position, &mirror_position);
 309       if (master_position == mirror_position) {
 310         break;
 311       }
 312       wait_for_watcher_notify(1);
 313     }
 314
 315     ASSERT_EQ(master_position, mirror_position);
 316   }
 317
 318   void wait_for_stopped() {
 319     for (int i = 0; i < 100; i++) {
 320       if (m_replayer->is_stopped()) {
 321         break;
 322       }
 323       wait_for_watcher_notify(1);
 324     }
 325     ASSERT_TRUE(m_replayer->is_stopped());
 326   }
 327
 328   void write_test_data(librbd::ImageCtx *ictx, const char *test_data, off_t off,
 329                        size_t len)
 330   {
 331     size_t written;
 332     bufferlist bl;
 333     bl.append(std::string(test_data, len));
 334     written = ictx->io_work_queue->write(off, len, std::move(bl), 0);
 335     printf("wrote: %d\n", (int)written);
 336     ASSERT_EQ(len, written);
 337   }
 338
 339   void read_test_data(librbd::ImageCtx *ictx, const char *expected, off_t off,
 340                       size_t len)
 341   {
 342     ssize_t read;
 343     char *result = (char *)malloc(len + 1);
 344
 345     ASSERT_NE(static_cast<char *>(NULL), result);
 346     read = ictx->io_work_queue->read(
 347       off, len, librbd::io::ReadResult{result, len}, 0);
 348     printf("read: %d\n", (int)read);
 349     ASSERT_EQ(len, static_cast<size_t>(read));
 350     result[len] = '\0';
 351     if (memcmp(result, expected, len)) {
 352       printf("read: %s\nexpected: %s\n", result, expected);
 353       ASSERT_EQ(0, memcmp(result, expected, len));
 354     }
 355     free(result);
 356   }
 357
 358   void generate_test_data() {
 359     for (int i = 0; i < TEST_IO_SIZE; ++i) {
 360       m_test_data[i] = (char) (rand() % (126 - 33) + 33);
 361     }
 362     m_test_data[TEST_IO_SIZE] = '\0';
 363   }
 364
 365   void flush(librbd::ImageCtx *ictx)
 366   {
 367     C_SaferCond aio_flush_ctx;
 368     auto c = librbd::io::AioCompletion::create(&aio_flush_ctx);
 369     c->get();
 370     ictx->io_work_queue->aio_flush(c);
 371     ASSERT_EQ(0, c->wait_for_complete());
 372     c->put();
 373
 374     C_SaferCond journal_flush_ctx;
 375     ictx->journal->flush_commit_position(&journal_flush_ctx);
 376     ASSERT_EQ(0, journal_flush_ctx.wait());
 377
 378     printf("flushed\n");
 379   }
 380
 381   static int _image_number;
 382
 383   std::shared_ptr<librados::Rados> m_local_cluster;
 384   std::unique_ptr<rbd::mirror::Threads<>> m_threads;
 385   std::unique_ptr<rbd::mirror::ServiceDaemon<>> m_service_daemon;
 386   std::unique_ptr<rbd::mirror::ImageDeleter<>> m_image_deleter;
 387   librados::Rados m_remote_cluster;
 388   rbd::mirror::InstanceWatcher<> *m_instance_watcher;
 389   std::string m_local_mirror_uuid = "local mirror uuid";
 390   std::string m_remote_mirror_uuid = "remote mirror uuid";
 391   std::string m_local_pool_name, m_remote_pool_name;
 392   librados::IoCtx m_local_ioctx, m_remote_ioctx;
 393   std::string m_image_name;
 394   int64_t m_remote_pool_id;
 395   std::string m_remote_image_id;
 396   std::string m_global_image_id;
 397   rbd::mirror::ImageReplayer<> *m_replayer;
 398   C_WatchCtx *m_watch_ctx;
 399   uint64_t m_watch_handle;
 400   char m_test_data[TEST_IO_SIZE + 1];
 401 };
 402
 403 int TestImageReplayer::_image_number;
 404
 405 TEST_F(TestImageReplayer, Bootstrap)
 406 {
 407   bootstrap();
 408 }
 409
 410 TEST_F(TestImageReplayer, BootstrapErrorLocalImageExists)
 411 {
 412   int order = 0;
 413   EXPECT_EQ(0, librbd::create(m_local_ioctx, m_image_name.c_str(), 1 << 22,
 414                               false, 0, &order, 0, 0));
 415
 416   create_replayer<>();
 417   C_SaferCond cond;
 418   m_replayer->start(&cond);
 419   ASSERT_EQ(-EEXIST, cond.wait());
 420 }
 421
 422 TEST_F(TestImageReplayer, BootstrapErrorNoJournal)
 423 {
 424   ASSERT_EQ(0, librbd::Journal<>::remove(m_remote_ioctx, m_remote_image_id));
 425
 426   create_replayer<>();
 427   C_SaferCond cond;
 428   m_replayer->start(&cond);
 429   ASSERT_EQ(-ENOENT, cond.wait());
 430 }
 431
 432 TEST_F(TestImageReplayer, BootstrapErrorMirrorDisabled)
 433 {
 434   // disable remote image mirroring
 435   ASSERT_EQ(0, librbd::api::Mirror<>::mode_set(m_remote_ioctx,
 436                                                RBD_MIRROR_MODE_IMAGE));
 437   librbd::ImageCtx *ictx;
 438   open_remote_image(&ictx);
 439   ASSERT_EQ(0, librbd::api::Mirror<>::image_disable(ictx, true));
 440   close_image(ictx);
 441
 442   create_replayer<>();
 443   C_SaferCond cond;
 444   m_replayer->start(&cond);
 445   ASSERT_EQ(-ENOENT, cond.wait());
 446 }
 447
 448 TEST_F(TestImageReplayer, BootstrapMirrorDisabling)
 449 {
 450   // set remote image mirroring state to DISABLING
 451   ASSERT_EQ(0, librbd::api::Mirror<>::mode_set(m_remote_ioctx,
 452                                                RBD_MIRROR_MODE_IMAGE));
 453   librbd::ImageCtx *ictx;
 454   open_remote_image(&ictx);
 455   ASSERT_EQ(0, librbd::api::Mirror<>::image_enable(ictx, false));
 456   cls::rbd::MirrorImage mirror_image;
 457   ASSERT_EQ(0, librbd::cls_client::mirror_image_get(&m_remote_ioctx, ictx->id,
 458                                                     &mirror_image));
 459   mirror_image.state = cls::rbd::MirrorImageState::MIRROR_IMAGE_STATE_DISABLING;
 460   ASSERT_EQ(0, librbd::cls_client::mirror_image_set(&m_remote_ioctx, ictx->id,
 461                                                     mirror_image));
 462   close_image(ictx);
 463
 464   create_replayer<>();
 465   C_SaferCond cond;
 466   m_replayer->start(&cond);
 467   ASSERT_EQ(-EREMOTEIO, cond.wait());
 468   ASSERT_TRUE(m_replayer->is_stopped());
 469 }
 470
 471 TEST_F(TestImageReplayer, BootstrapDemoted)
 472 {
 473   // demote remote image
 474   librbd::ImageCtx *ictx;
 475   open_remote_image(&ictx);
 476   ASSERT_EQ(0, librbd::api::Mirror<>::image_demote(ictx));
 477   close_image(ictx);
 478
 479   create_replayer<>();
 480   C_SaferCond cond;
 481   m_replayer->start(&cond);
 482   ASSERT_EQ(-EREMOTEIO, cond.wait());
 483   ASSERT_TRUE(m_replayer->is_stopped());
 484 }
 485
 486 TEST_F(TestImageReplayer, StartInterrupted)
 487 {
 488   create_replayer<>();
 489   C_SaferCond start_cond, stop_cond;
 490   m_replayer->start(&start_cond);
 491   m_replayer->stop(&stop_cond);
 492   int r = start_cond.wait();
 493   printf("start returned %d\n", r);
 494   // TODO: improve the test to avoid this race
 495   ASSERT_TRUE(r == -ECANCELED || r == 0);
 496   ASSERT_EQ(0, stop_cond.wait());
 497 }
 498
 499 TEST_F(TestImageReplayer, JournalReset)
 500 {
 501   bootstrap();
 502   delete m_replayer;
 503
 504   ASSERT_EQ(0, librbd::Journal<>::reset(m_remote_ioctx, m_remote_image_id));
 505
 506   // try to recover
 507   bootstrap();
 508 }
 509
 510 TEST_F(TestImageReplayer, ErrorNoJournal)
 511 {
 512   bootstrap();
 513
 514   // disable remote journal journaling
 515   // (reset before disabling, so it does not fail with EBUSY)
 516   ASSERT_EQ(0, librbd::Journal<>::reset(m_remote_ioctx, m_remote_image_id));
 517   librbd::ImageCtx *ictx;
 518   open_remote_image(&ictx);
 519   uint64_t features;
 520   ASSERT_EQ(0, librbd::get_features(ictx, &features));
 521   ASSERT_EQ(0, ictx->operations->update_features(RBD_FEATURE_JOURNALING,
 522                                                  false));
 523   close_image(ictx);
 524
 525   C_SaferCond cond;
 526   m_replayer->start(&cond);
 527   ASSERT_EQ(0, cond.wait());
 528 }
 529
 530 TEST_F(TestImageReplayer, StartStop)
 531 {
 532   bootstrap();
 533
 534   start();
 535   wait_for_replay_complete();
 536   stop();
 537 }
 538
 539 TEST_F(TestImageReplayer, WriteAndStartReplay)
 540 {
 541   bootstrap();
 542
 543   // Write to remote image and start replay
 544
 545   librbd::ImageCtx *ictx;
 546
 547   generate_test_data();
 548   open_remote_image(&ictx);
 549   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 550     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 551   }
 552   flush(ictx);
 553   close_image(ictx);
 554
 555   start();
 556   wait_for_replay_complete();
 557   stop();
 558
 559   open_local_image(&ictx);
 560   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 561     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 562   }
 563   close_image(ictx);
 564 }
 565
 566 TEST_F(TestImageReplayer, StartReplayAndWrite)
 567 {
 568   bootstrap();
 569
 570   // Start replay and write to remote image
 571
 572   librbd::ImageCtx *ictx;
 573
 574   start();
 575
 576   generate_test_data();
 577   open_remote_image(&ictx);
 578   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 579     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 580   }
 581   flush(ictx);
 582
 583   wait_for_replay_complete();
 584
 585   for (int i = TEST_IO_COUNT; i < 2 * TEST_IO_COUNT; ++i) {
 586     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 587   }
 588   flush(ictx);
 589   close_image(ictx);
 590
 591   wait_for_replay_complete();
 592
 593   open_local_image(&ictx);
 594   for (int i = 0; i < 2 * TEST_IO_COUNT; ++i) {
 595     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 596   }
 597   close_image(ictx);
 598
 599   stop();
 600 }
 601
 602 TEST_F(TestImageReplayer, NextTag)
 603 {
 604   bootstrap();
 605
 606   // write, reopen, and write again to test switch to the next tag
 607
 608   librbd::ImageCtx *ictx;
 609
 610   start();
 611
 612   generate_test_data();
 613
 614   const int N = 10;
 615
 616   for (int j = 0; j < N; j++) {
 617     open_remote_image(&ictx);
 618     for (int i = j * TEST_IO_COUNT; i < (j + 1) * TEST_IO_COUNT; ++i) {
 619       write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 620     }
 621     close_image(ictx);
 622   }
 623
 624   wait_for_replay_complete();
 625
 626   open_local_image(&ictx);
 627   for (int i = 0; i < N * TEST_IO_COUNT; ++i) {
 628     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 629   }
 630   close_image(ictx);
 631
 632   stop();
 633 }
 634
 635 TEST_F(TestImageReplayer, Resync)
 636 {
 637   bootstrap();
 638
 639   librbd::ImageCtx *ictx;
 640
 641   start();
 642
 643   generate_test_data();
 644
 645   open_remote_image(&ictx);
 646   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 647     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 648   }
 649   flush(ictx);
 650
 651   wait_for_replay_complete();
 652
 653   for (int i = TEST_IO_COUNT; i < 2 * TEST_IO_COUNT; ++i) {
 654     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 655   }
 656   flush(ictx);
 657   close_image(ictx);
 658
 659   C_SaferCond ctx;
 660   m_replayer->resync_image(&ctx);
 661   ASSERT_EQ(0, ctx.wait());
 662
 663   C_SaferCond delete_ctx;
 664   m_image_deleter->wait_for_scheduled_deletion(
 665     m_local_ioctx.get_id(), m_replayer->get_global_image_id(), &delete_ctx);
 666   EXPECT_EQ(0, delete_ctx.wait());
 667
 668   C_SaferCond cond;
 669   m_replayer->start(&cond);
 670   ASSERT_EQ(0, cond.wait());
 671
 672   ASSERT_TRUE(m_replayer->is_replaying());
 673
 674   wait_for_replay_complete();
 675
 676   open_local_image(&ictx);
 677   for (int i = 0; i < 2 * TEST_IO_COUNT; ++i) {
 678     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 679   }
 680   close_image(ictx);
 681
 682   stop();
 683 }
 684
 685 TEST_F(TestImageReplayer, Resync_While_Stop)
 686 {
 687
 688   bootstrap();
 689
 690   start();
 691
 692   generate_test_data();
 693
 694   librbd::ImageCtx *ictx;
 695   open_remote_image(&ictx);
 696   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 697     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 698   }
 699   flush(ictx);
 700
 701   wait_for_replay_complete();
 702
 703   for (int i = TEST_IO_COUNT; i < 2 * TEST_IO_COUNT; ++i) {
 704     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 705   }
 706   flush(ictx);
 707   close_image(ictx);
 708
 709   wait_for_replay_complete();
 710
 711   C_SaferCond cond;
 712   m_replayer->stop(&cond);
 713   ASSERT_EQ(0, cond.wait());
 714
 715   open_local_image(&ictx);
 716   librbd::Journal<>::request_resync(ictx);
 717   close_image(ictx);
 718
 719   C_SaferCond cond2;
 720   m_replayer->start(&cond2);
 721   ASSERT_EQ(0, cond2.wait());
 722
 723   ASSERT_TRUE(m_replayer->is_stopped());
 724
 725   C_SaferCond delete_ctx;
 726   m_image_deleter->wait_for_scheduled_deletion(
 727     m_local_ioctx.get_id(), m_replayer->get_global_image_id(), &delete_ctx);
 728   EXPECT_EQ(0, delete_ctx.wait());
 729
 730   C_SaferCond cond3;
 731   m_replayer->start(&cond3);
 732   ASSERT_EQ(0, cond3.wait());
 733
 734   ASSERT_TRUE(m_replayer->is_replaying());
 735
 736   wait_for_replay_complete();
 737
 738   open_local_image(&ictx);
 739   for (int i = 0; i < 2 * TEST_IO_COUNT; ++i) {
 740     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 741   }
 742   close_image(ictx);
 743
 744   stop();
 745 }
 746
 747 TEST_F(TestImageReplayer, Resync_StartInterrupted)
 748 {
 749
 750   bootstrap();
 751
 752   librbd::ImageCtx *ictx;
 753   open_local_image(&ictx);
 754   librbd::Journal<>::request_resync(ictx);
 755   close_image(ictx);
 756
 757   C_SaferCond cond;
 758   m_replayer->start(&cond);
 759   ASSERT_EQ(0, cond.wait());
 760
 761   ASSERT_TRUE(m_replayer->is_stopped());
 762
 763   C_SaferCond delete_ctx;
 764   m_image_deleter->wait_for_scheduled_deletion(
 765     m_local_ioctx.get_id(), m_replayer->get_global_image_id(), &delete_ctx);
 766   EXPECT_EQ(0, delete_ctx.wait());
 767
 768   C_SaferCond cond2;
 769   m_replayer->start(&cond2);
 770   ASSERT_EQ(0, cond2.wait());
 771
 772   ASSERT_EQ(0U, m_watch_handle);
 773   std::string oid = ::journal::Journaler::header_oid(m_remote_image_id);
 774   m_watch_ctx = new C_WatchCtx(this, oid);
 775   ASSERT_EQ(0, m_remote_ioctx.watch2(oid, &m_watch_handle, m_watch_ctx));
 776
 777   ASSERT_TRUE(m_replayer->is_replaying());
 778
 779   generate_test_data();
 780   open_remote_image(&ictx);
 781   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 782     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 783   }
 784   flush(ictx);
 785
 786   wait_for_replay_complete();
 787
 788   for (int i = TEST_IO_COUNT; i < 2 * TEST_IO_COUNT; ++i) {
 789     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 790   }
 791   flush(ictx);
 792   close_image(ictx);
 793
 794   wait_for_replay_complete();
 795
 796   open_local_image(&ictx);
 797   for (int i = 0; i < 2 * TEST_IO_COUNT; ++i) {
 798     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 799   }
 800   close_image(ictx);
 801
 802   stop();
 803 }
 804
 805 TEST_F(TestImageReplayer, MultipleReplayFailures_SingleEpoch) {
 806   bootstrap();
 807
 808   // inject a snapshot that cannot be unprotected
 809   librbd::ImageCtx *ictx;
 810   open_image(m_local_ioctx, m_image_name, false, &ictx);
 811   ictx->features &= ~RBD_FEATURE_JOURNALING;
 812   ASSERT_EQ(0, ictx->operations->snap_create(cls::rbd::UserSnapshotNamespace(),
 813                                              "foo"));
 814   ASSERT_EQ(0, ictx->operations->snap_protect(cls::rbd::UserSnapshotNamespace(),
 815                                               "foo"));
 816   ASSERT_EQ(0, librbd::cls_client::add_child(&ictx->md_ctx, RBD_CHILDREN,
 817                                              {ictx->md_ctx.get_id(),
 818                                               ictx->id,
 819                                               ictx->snap_ids[{cls::rbd::UserSnapshotNamespace(), "foo"}]},
 820                                              "dummy child id"));
 821   close_image(ictx);
 822
 823   // race failed op shut down with new ops
 824   open_remote_image(&ictx);
 825   for (uint64_t i = 0; i < 10; ++i) {
 826     RWLock::RLocker owner_locker(ictx->owner_lock);
 827     C_SaferCond request_lock;
 828     ictx->exclusive_lock->acquire_lock(&request_lock);
 829     ASSERT_EQ(0, request_lock.wait());
 830
 831     C_SaferCond append_ctx;
 832     ictx->journal->append_op_event(
 833       i,
 834       librbd::journal::EventEntry{
 835         librbd::journal::SnapUnprotectEvent{i,
 836                                             cls::rbd::UserSnapshotNamespace(),
 837                                             "foo"}},
 838       &append_ctx);
 839     ASSERT_EQ(0, append_ctx.wait());
 840
 841     C_SaferCond commit_ctx;
 842     ictx->journal->commit_op_event(i, 0, &commit_ctx);
 843     ASSERT_EQ(0, commit_ctx.wait());
 844
 845     C_SaferCond release_ctx;
 846     ictx->exclusive_lock->release_lock(&release_ctx);
 847     ASSERT_EQ(0, release_ctx.wait());
 848   }
 849
 850   for (uint64_t i = 0; i < 5; ++i) {
 851     start();
 852     wait_for_stopped();
 853     unwatch();
 854   }
 855   close_image(ictx);
 856 }
 857
 858 TEST_F(TestImageReplayer, MultipleReplayFailures_MultiEpoch) {
 859   bootstrap();
 860
 861   // inject a snapshot that cannot be unprotected
 862   librbd::ImageCtx *ictx;
 863   open_image(m_local_ioctx, m_image_name, false, &ictx);
 864   ictx->features &= ~RBD_FEATURE_JOURNALING;
 865   ASSERT_EQ(0, ictx->operations->snap_create(cls::rbd::UserSnapshotNamespace(),
 866                                              "foo"));
 867   ASSERT_EQ(0, ictx->operations->snap_protect(cls::rbd::UserSnapshotNamespace(),
 868                                               "foo"));
 869   ASSERT_EQ(0, librbd::cls_client::add_child(&ictx->md_ctx, RBD_CHILDREN,
 870                                              {ictx->md_ctx.get_id(),
 871                                               ictx->id,
 872                                               ictx->snap_ids[{cls::rbd::UserSnapshotNamespace(),
 873                                                               "foo"}]},
 874                                              "dummy child id"));
 875   close_image(ictx);
 876
 877   // race failed op shut down with new tag flush
 878   open_remote_image(&ictx);
 879   {
 880     RWLock::RLocker owner_locker(ictx->owner_lock);
 881     C_SaferCond request_lock;
 882     ictx->exclusive_lock->acquire_lock(&request_lock);
 883     ASSERT_EQ(0, request_lock.wait());
 884
 885     C_SaferCond append_ctx;
 886     ictx->journal->append_op_event(
 887       1U,
 888       librbd::journal::EventEntry{
 889         librbd::journal::SnapUnprotectEvent{1U,
 890                                             cls::rbd::UserSnapshotNamespace(),
 891                                             "foo"}},
 892       &append_ctx);
 893     ASSERT_EQ(0, append_ctx.wait());
 894
 895     C_SaferCond commit_ctx;
 896     ictx->journal->commit_op_event(1U, 0, &commit_ctx);
 897     ASSERT_EQ(0, commit_ctx.wait());
 898
 899     C_SaferCond release_ctx;
 900     ictx->exclusive_lock->release_lock(&release_ctx);
 901     ASSERT_EQ(0, release_ctx.wait());
 902   }
 903
 904   generate_test_data();
 905   write_test_data(ictx, m_test_data, 0, TEST_IO_SIZE);
 906
 907   for (uint64_t i = 0; i < 5; ++i) {
 908     start();
 909     wait_for_stopped();
 910     unwatch();
 911   }
 912   close_image(ictx);
 913 }
 914
 915 TEST_F(TestImageReplayer, Disconnect)
 916 {
 917   bootstrap();
 918
 919   // Make sure rbd_mirroring_resync_after_disconnect is not set
 920   EXPECT_EQ(0, m_local_cluster->conf_set("rbd_mirroring_resync_after_disconnect", "false"));
 921
 922   // Test start fails if disconnected
 923
 924   librbd::ImageCtx *ictx;
 925
 926   generate_test_data();
 927   open_remote_image(&ictx);
 928   for (int i = 0; i < TEST_IO_COUNT; ++i) {
 929     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 930   }
 931   flush(ictx);
 932   close_image(ictx);
 933
 934   std::string oid = ::journal::Journaler::header_oid(m_remote_image_id);
 935   ASSERT_EQ(0, cls::journal::client::client_update_state(m_remote_ioctx, oid,
 936         m_local_mirror_uuid, cls::journal::CLIENT_STATE_DISCONNECTED));
 937
 938   C_SaferCond cond1;
 939   m_replayer->start(&cond1);
 940   ASSERT_EQ(-ENOTCONN, cond1.wait());
 941
 942   // Test start succeeds after resync
 943
 944   open_local_image(&ictx);
 945   librbd::Journal<>::request_resync(ictx);
 946   close_image(ictx);
 947   C_SaferCond cond2;
 948   m_replayer->start(&cond2);
 949   ASSERT_EQ(0, cond2.wait());
 950   C_SaferCond delete_cond;
 951   m_image_deleter->wait_for_scheduled_deletion(
 952     m_local_ioctx.get_id(), m_replayer->get_global_image_id(), &delete_cond);
 953   EXPECT_EQ(0, delete_cond.wait());
 954
 955   start();
 956   wait_for_replay_complete();
 957
 958   // Test replay stopped after disconnect
 959
 960   open_remote_image(&ictx);
 961   for (int i = TEST_IO_COUNT; i < 2 * TEST_IO_COUNT; ++i) {
 962     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
 963   }
 964   flush(ictx);
 965   close_image(ictx);
 966
 967   ASSERT_EQ(0, cls::journal::client::client_update_state(m_remote_ioctx, oid,
 968         m_local_mirror_uuid, cls::journal::CLIENT_STATE_DISCONNECTED));
 969   bufferlist bl;
 970   ASSERT_EQ(0, m_remote_ioctx.notify2(oid, bl, 5000, NULL));
 971
 972   wait_for_stopped();
 973
 974   // Test start fails after disconnect
 975
 976   C_SaferCond cond3;
 977   m_replayer->start(&cond3);
 978   ASSERT_EQ(-ENOTCONN, cond3.wait());
 979   C_SaferCond cond4;
 980   m_replayer->start(&cond4);
 981   ASSERT_EQ(-ENOTCONN, cond4.wait());
 982
 983   // Test automatic resync if rbd_mirroring_resync_after_disconnect is set
 984
 985   EXPECT_EQ(0, m_local_cluster->conf_set("rbd_mirroring_resync_after_disconnect", "true"));
 986
 987   // Resync is flagged on first start attempt
 988   C_SaferCond cond5;
 989   m_replayer->start(&cond5);
 990   ASSERT_EQ(-ENOTCONN, cond5.wait());
 991   C_SaferCond delete_cond1;
 992   m_image_deleter->wait_for_scheduled_deletion(
 993     m_local_ioctx.get_id(), m_replayer->get_global_image_id(), &delete_cond1);
 994   EXPECT_EQ(0, delete_cond1.wait());
 995
 996   C_SaferCond cond6;
 997   m_replayer->start(&cond6);
 998   ASSERT_EQ(0, cond6.wait());
 999   wait_for_replay_complete();
1000
1001   stop();
1002 }
1003
1004 TEST_F(TestImageReplayer, UpdateFeatures)
1005 {
1006   const uint64_t FEATURES_TO_UPDATE =
1007     RBD_FEATURE_OBJECT_MAP | RBD_FEATURE_FAST_DIFF;
1008
1009   uint64_t features;
1010   librbd::ImageCtx *ictx;
1011
1012   // Make sure the features we will update are disabled initially
1013
1014   open_remote_image(&ictx);
1015   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1016   features &= FEATURES_TO_UPDATE;
1017   if (features) {
1018     ASSERT_EQ(0, ictx->operations->update_features(FEATURES_TO_UPDATE,
1019                                                    false));
1020   }
1021   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1022   ASSERT_EQ(0U, features & FEATURES_TO_UPDATE);
1023   close_image(ictx);
1024
1025   bootstrap();
1026
1027   open_remote_image(&ictx);
1028   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1029   ASSERT_EQ(0U, features & FEATURES_TO_UPDATE);
1030   close_image(ictx);
1031
1032   open_local_image(&ictx);
1033   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1034   ASSERT_EQ(0U, features & FEATURES_TO_UPDATE);
1035   close_image(ictx);
1036
1037   // Start replay and update features
1038
1039   start();
1040
1041   open_remote_image(&ictx);
1042   ASSERT_EQ(0, ictx->operations->update_features(FEATURES_TO_UPDATE,
1043                                                  true));
1044   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1045   ASSERT_EQ(FEATURES_TO_UPDATE, features & FEATURES_TO_UPDATE);
1046   close_image(ictx);
1047
1048   wait_for_replay_complete();
1049
1050   open_local_image(&ictx);
1051   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1052   ASSERT_EQ(FEATURES_TO_UPDATE, features & FEATURES_TO_UPDATE);
1053   close_image(ictx);
1054
1055   open_remote_image(&ictx);
1056   ASSERT_EQ(0, ictx->operations->update_features(FEATURES_TO_UPDATE,
1057                                                  false));
1058   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1059   ASSERT_EQ(0U, features & FEATURES_TO_UPDATE);
1060   close_image(ictx);
1061
1062   wait_for_replay_complete();
1063
1064   open_local_image(&ictx);
1065   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1066   ASSERT_EQ(0U, features & FEATURES_TO_UPDATE);
1067   close_image(ictx);
1068
1069   // Test update_features error does not stop replication
1070
1071   open_remote_image(&ictx);
1072   ASSERT_EQ(0, librbd::get_features(ictx, &features));
1073   ASSERT_NE(0U, features & RBD_FEATURE_EXCLUSIVE_LOCK);
1074   ASSERT_EQ(-EINVAL, ictx->operations->update_features(RBD_FEATURE_EXCLUSIVE_LOCK,
1075                                                        false));
1076   generate_test_data();
1077   for (int i = 0; i < TEST_IO_COUNT; ++i) {
1078     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
1079   }
1080   flush(ictx);
1081   close_image(ictx);
1082
1083   wait_for_replay_complete();
1084
1085   open_local_image(&ictx);
1086   for (int i = 0; i < TEST_IO_COUNT; ++i) {
1087     read_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
1088   }
1089   close_image(ictx);
1090
1091   stop();
1092 }
1093
1094 TEST_F(TestImageReplayer, MetadataSetRemove)
1095 {
1096   const std::string KEY = "test_key";
1097   const std::string VALUE = "test_value";
1098
1099   librbd::ImageCtx *ictx;
1100   std::string value;
1101
1102   bootstrap();
1103
1104   start();
1105
1106   // Test metadata_set replication
1107
1108   open_remote_image(&ictx);
1109   ASSERT_EQ(0, ictx->operations->metadata_set(KEY, VALUE));
1110   value.clear();
1111   ASSERT_EQ(0, librbd::metadata_get(ictx, KEY, &value));
1112   ASSERT_EQ(VALUE, value);
1113   close_image(ictx);
1114
1115   wait_for_replay_complete();
1116
1117   open_local_image(&ictx);
1118   value.clear();
1119   ASSERT_EQ(0, librbd::metadata_get(ictx, KEY, &value));
1120   ASSERT_EQ(VALUE, value);
1121   close_image(ictx);
1122
1123   // Test metadata_remove replication
1124
1125   open_remote_image(&ictx);
1126   ASSERT_EQ(0, ictx->operations->metadata_remove(KEY));
1127   ASSERT_EQ(-ENOENT, librbd::metadata_get(ictx, KEY, &value));
1128   close_image(ictx);
1129
1130   wait_for_replay_complete();
1131
1132   open_local_image(&ictx);
1133   ASSERT_EQ(-ENOENT, librbd::metadata_get(ictx, KEY, &value));
1134   close_image(ictx);
1135
1136   stop();
1137 }
1138
1139 TEST_F(TestImageReplayer, MirroringDelay)
1140 {
1141   const double DELAY = 10; // set less than wait_for_replay_complete timeout
1142
1143   librbd::ImageCtx *ictx;
1144   utime_t start_time;
1145   double delay;
1146
1147   bootstrap();
1148
1149   ASSERT_EQ(0, m_local_cluster->conf_set("rbd_mirroring_replay_delay",
1150                                          stringify(DELAY).c_str()));
1151   open_local_image(&ictx);
1152   ASSERT_EQ(DELAY, ictx->mirroring_replay_delay);
1153   close_image(ictx);
1154
1155   start();
1156
1157   // Test delay
1158
1159   generate_test_data();
1160   open_remote_image(&ictx);
1161   start_time = ceph_clock_now();
1162   for (int i = 0; i < TEST_IO_COUNT; ++i) {
1163     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
1164   }
1165   flush(ictx);
1166   close_image(ictx);
1167
1168   wait_for_replay_complete();
1169   delay = ceph_clock_now() - start_time;
1170   ASSERT_GE(delay, DELAY);
1171
1172   // Test stop when delaying replay
1173
1174   open_remote_image(&ictx);
1175   start_time = ceph_clock_now();
1176   for (int i = 0; i < TEST_IO_COUNT; ++i) {
1177     write_test_data(ictx, m_test_data, TEST_IO_SIZE * i, TEST_IO_SIZE);
1178   }
1179   close_image(ictx);
1180
1181   sleep(DELAY / 2);
1182   stop();
1183   start();
1184
1185   wait_for_replay_complete();
1186   delay = ceph_clock_now() - start_time;
1187   ASSERT_GE(delay, DELAY);
1188
1189   stop();
1190 }