ceph/src/test/rbd_mirror/test_Instances.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 "include/rados/librados.hpp"
   5 #include "cls/rbd/cls_rbd_client.h"
   6 #include "test/rbd_mirror/test_fixture.h"
   7 #include "tools/rbd_mirror/InstanceWatcher.h"
   8 #include "tools/rbd_mirror/Instances.h"
   9 #include "tools/rbd_mirror/Threads.h"
  10 #include "common/Cond.h"
  11
  12 #include "test/librados/test.h"
  13 #include "gtest/gtest.h"
  14 #include <vector>
  15
  16 using rbd::mirror::InstanceWatcher;
  17 using rbd::mirror::Instances;
  18
  19 void register_test_instances() {
  20 }
  21
  22 class TestInstances : public ::rbd::mirror::TestFixture {
  23 public:
  24   struct Listener : public rbd::mirror::instances::Listener {
  25     std::mutex lock;
  26
  27     struct Instance {
  28       uint32_t count = 0;
  29       std::set<std::string> ids;
  30       C_SaferCond ctx;
  31     };
  32
  33     Instance add;
  34     Instance remove;
  35
  36     void handle(const InstanceIds& instance_ids, Instance* instance) {
  37       std::unique_lock<std::mutex> locker(lock);
  38       for (auto& instance_id : instance_ids) {
  39         ceph_assert(instance->count > 0);
  40         --instance->count;
  41
  42         instance->ids.insert(instance_id);
  43         if (instance->count == 0) {
  44           instance->ctx.complete(0);
  45         }
  46       }
  47     }
  48
  49     void handle_added(const InstanceIds& instance_ids) override {
  50       handle(instance_ids, &add);
  51     }
  52
  53     void handle_removed(const InstanceIds& instance_ids) override {
  54       handle(instance_ids, &remove);
  55     }
  56   };
  57
  58   virtual void SetUp() {
  59     TestFixture::SetUp();
  60     m_local_io_ctx.remove(RBD_MIRROR_LEADER);
  61     EXPECT_EQ(0, m_local_io_ctx.create(RBD_MIRROR_LEADER, true));
  62
  63     m_instance_id = stringify(m_local_io_ctx.get_instance_id());
  64   }
  65
  66   Listener m_listener;
  67   std::string m_instance_id;
  68 };
  69
  70 TEST_F(TestInstances, InitShutdown)
  71 {
  72   m_listener.add.count = 1;
  73   Instances<> instances(m_threads, m_local_io_ctx, m_instance_id, m_listener);
  74
  75   std::string instance_id = "instance_id";
  76   ASSERT_EQ(0, librbd::cls_client::mirror_instances_add(&m_local_io_ctx,
  77                                                         instance_id));
  78
  79   C_SaferCond on_init;
  80   instances.init(&on_init);
  81   ASSERT_EQ(0, on_init.wait());
  82
  83   ASSERT_LT(0U, m_listener.add.count);
  84   instances.unblock_listener();
  85
  86   ASSERT_EQ(0, m_listener.add.ctx.wait());
  87   ASSERT_EQ(std::set<std::string>({instance_id}), m_listener.add.ids);
  88
  89   C_SaferCond on_shut_down;
  90   instances.shut_down(&on_shut_down);
  91   ASSERT_EQ(0, on_shut_down.wait());
  92 }
  93
  94 TEST_F(TestInstances, InitEnoent)
  95 {
  96   Instances<> instances(m_threads, m_local_io_ctx, m_instance_id, m_listener);
  97
  98   m_local_io_ctx.remove(RBD_MIRROR_LEADER);
  99
 100   C_SaferCond on_init;
 101   instances.init(&on_init);
 102   ASSERT_EQ(0, on_init.wait());
 103
 104   C_SaferCond on_shut_down;
 105   instances.shut_down(&on_shut_down);
 106   ASSERT_EQ(0, on_shut_down.wait());
 107 }
 108
 109 TEST_F(TestInstances, NotifyRemove)
 110 {
 111   // speed testing up a little
 112   EXPECT_EQ(0, _rados->conf_set("rbd_mirror_leader_heartbeat_interval", "1"));
 113   EXPECT_EQ(0, _rados->conf_set("rbd_mirror_leader_max_missed_heartbeats",
 114                                 "2"));
 115   EXPECT_EQ(0, _rados->conf_set("rbd_mirror_leader_max_acquire_attempts_before_break",
 116                                 "0"));
 117
 118   m_listener.add.count = 2;
 119   m_listener.remove.count = 1;
 120   Instances<> instances(m_threads, m_local_io_ctx, m_instance_id, m_listener);
 121
 122   std::string instance_id1 = "instance_id1";
 123   std::string instance_id2 = "instance_id2";
 124
 125   ASSERT_EQ(0, librbd::cls_client::mirror_instances_add(&m_local_io_ctx,
 126                                                         instance_id1));
 127
 128   C_SaferCond on_init;
 129   instances.init(&on_init);
 130   ASSERT_EQ(0, on_init.wait());
 131
 132   instances.acked({instance_id2});
 133
 134   ASSERT_LT(0U, m_listener.add.count);
 135   instances.unblock_listener();
 136
 137   ASSERT_EQ(0, m_listener.add.ctx.wait());
 138   ASSERT_EQ(std::set<std::string>({instance_id1, instance_id2}),
 139             m_listener.add.ids);
 140
 141   std::vector<std::string> instance_ids;
 142   for (int i = 0; i < 100; i++) {
 143     instances.acked({instance_id1});
 144     if (m_listener.remove.count > 0) {
 145       usleep(250000);
 146     }
 147   }
 148
 149   instances.acked({instance_id1});
 150   ASSERT_EQ(0, m_listener.remove.ctx.wait());
 151   ASSERT_EQ(std::set<std::string>({instance_id2}),
 152            m_listener.remove.ids);
 153
 154   C_SaferCond on_get;
 155   instances.acked({instance_id1});
 156   InstanceWatcher<>::get_instances(m_local_io_ctx, &instance_ids, &on_get);
 157   EXPECT_EQ(0, on_get.wait());
 158   EXPECT_EQ(1U, instance_ids.size());
 159   ASSERT_EQ(instance_ids[0], instance_id1);
 160
 161   C_SaferCond on_shut_down;
 162   instances.shut_down(&on_shut_down);
 163   ASSERT_EQ(0, on_shut_down.wait());
 164 }