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