ceph/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.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 "ReplayStatusFormatter.h"
   5 #include "common/debug.h"
   6 #include "common/dout.h"
   7 #include "common/errno.h"
   8 #include "journal/Journaler.h"
   9 #include "librbd/ImageCtx.h"
  10 #include "librbd/Journal.h"
  11 #include "librbd/Utils.h"
  12
  13 #define dout_context g_ceph_context
  14 #define dout_subsys ceph_subsys_rbd_mirror
  15 #undef dout_prefix
  16 #define dout_prefix *_dout << "rbd::mirror::image_replayer::ReplayStatusFormatter: " \
  17     << this << " " << __func__ << ": "
  18
  19 namespace rbd {
  20 namespace mirror {
  21 namespace image_replayer {
  22
  23 using librbd::util::unique_lock_name;
  24
  25 template <typename I>
  26 ReplayStatusFormatter<I>::ReplayStatusFormatter(Journaler *journaler,
  27                                                 const std::string &mirror_uuid)
  28   : m_journaler(journaler),
  29     m_mirror_uuid(mirror_uuid),
  30     m_lock(unique_lock_name("ReplayStatusFormatter::m_lock", this)) {
  31 }
  32
  33 template <typename I>
  34 bool ReplayStatusFormatter<I>::get_or_send_update(std::string *description,
  35                                                   Context *on_finish) {
  36   dout(20) << dendl;
  37
  38   bool in_progress = false;
  39   {
  40     Mutex::Locker locker(m_lock);
  41     if (m_on_finish) {
  42       in_progress = true;
  43     } else {
  44       m_on_finish = on_finish;
  45     }
  46   }
  47
  48   if (in_progress) {
  49     dout(10) << "previous request is still in progress, ignoring" << dendl;
  50     on_finish->complete(-EAGAIN);
  51     return false;
  52   }
  53
  54   m_master_position = cls::journal::ObjectPosition();
  55   m_mirror_position = cls::journal::ObjectPosition();
  56
  57   cls::journal::Client master_client, mirror_client;
  58   int r;
  59
  60   r = m_journaler->get_cached_client(librbd::Journal<>::IMAGE_CLIENT_ID,
  61                                      &master_client);
  62   if (r < 0) {
  63     derr << "error retrieving registered master client: "
  64          << cpp_strerror(r) << dendl;
  65   } else {
  66     r = m_journaler->get_cached_client(m_mirror_uuid, &mirror_client);
  67     if (r < 0) {
  68       derr << "error retrieving registered mirror client: "
  69            << cpp_strerror(r) << dendl;
  70     }
  71   }
  72
  73   if (!master_client.commit_position.object_positions.empty()) {
  74     m_master_position =
  75       *(master_client.commit_position.object_positions.begin());
  76   }
  77
  78   if (!mirror_client.commit_position.object_positions.empty()) {
  79     m_mirror_position =
  80       *(mirror_client.commit_position.object_positions.begin());
  81   }
  82
  83   if (!calculate_behind_master_or_send_update()) {
  84     dout(20) << "need to update tag cache" << dendl;
  85     return false;
  86   }
  87
  88   format(description);
  89
  90   {
  91     Mutex::Locker locker(m_lock);
  92     assert(m_on_finish == on_finish);
  93     m_on_finish = nullptr;
  94   }
  95
  96   on_finish->complete(-EEXIST);
  97   return true;
  98 }
  99
 100 template <typename I>
 101 bool ReplayStatusFormatter<I>::calculate_behind_master_or_send_update() {
 102   dout(20) << "m_master_position=" << m_master_position
 103            << ", m_mirror_position=" << m_mirror_position << dendl;
 104
 105   m_entries_behind_master = 0;
 106
 107   if (m_master_position == cls::journal::ObjectPosition() ||
 108       m_master_position.tag_tid < m_mirror_position.tag_tid) {
 109     return true;
 110   }
 111
 112   cls::journal::ObjectPosition master = m_master_position;
 113   uint64_t mirror_tag_tid = m_mirror_position.tag_tid;
 114
 115   while (master.tag_tid != mirror_tag_tid) {
 116     auto tag_it = m_tag_cache.find(master.tag_tid);
 117     if (tag_it == m_tag_cache.end()) {
 118       send_update_tag_cache(master.tag_tid, mirror_tag_tid);
 119       return false;
 120     }
 121     librbd::journal::TagData &tag_data = tag_it->second;
 122     m_entries_behind_master += master.entry_tid;
 123     master = cls::journal::ObjectPosition(0, tag_data.predecessor.tag_tid,
 124                                           tag_data.predecessor.entry_tid);
 125   }
 126   m_entries_behind_master += master.entry_tid - m_mirror_position.entry_tid;
 127
 128   dout(20) << "clearing tags not needed any more (below mirror position)"
 129            << dendl;
 130
 131   uint64_t tag_tid = mirror_tag_tid;
 132   size_t old_size = m_tag_cache.size();
 133   while (tag_tid != 0) {
 134     auto tag_it = m_tag_cache.find(tag_tid);
 135     if (tag_it == m_tag_cache.end()) {
 136       break;
 137     }
 138     librbd::journal::TagData &tag_data = tag_it->second;
 139
 140     dout(20) << "erasing tag " <<  tag_data << "for tag_tid " << tag_tid
 141              << dendl;
 142
 143     tag_tid = tag_data.predecessor.tag_tid;
 144     m_tag_cache.erase(tag_it);
 145   }
 146
 147   dout(20) << old_size - m_tag_cache.size() << " entries cleared" << dendl;
 148
 149   return true;
 150 }
 151
 152 template <typename I>
 153 void ReplayStatusFormatter<I>::send_update_tag_cache(uint64_t master_tag_tid,
 154                                                      uint64_t mirror_tag_tid) {
 155
 156   dout(20) << "master_tag_tid=" << master_tag_tid << ", mirror_tag_tid="
 157            << mirror_tag_tid << dendl;
 158
 159   if (master_tag_tid == mirror_tag_tid) {
 160     Context *on_finish = nullptr;
 161     {
 162       Mutex::Locker locker(m_lock);
 163       std::swap(m_on_finish, on_finish);
 164     }
 165
 166     assert(on_finish);
 167     on_finish->complete(0);
 168     return;
 169   }
 170
 171   FunctionContext *ctx = new FunctionContext(
 172     [this, master_tag_tid, mirror_tag_tid](int r) {
 173       handle_update_tag_cache(master_tag_tid, mirror_tag_tid, r);
 174     });
 175   m_journaler->get_tag(master_tag_tid, &m_tag, ctx);
 176 }
 177
 178 template <typename I>
 179 void ReplayStatusFormatter<I>::handle_update_tag_cache(uint64_t master_tag_tid,
 180                                                        uint64_t mirror_tag_tid,
 181                                                        int r) {
 182   librbd::journal::TagData tag_data;
 183
 184   if (r < 0) {
 185     derr << "error retrieving tag " << master_tag_tid << ": " << cpp_strerror(r)
 186          << dendl;
 187   } else {
 188     dout(20) << "retrieved tag " << master_tag_tid << ": " << m_tag << dendl;
 189
 190     bufferlist::iterator it = m_tag.data.begin();
 191     try {
 192       ::decode(tag_data, it);
 193     } catch (const buffer::error &err) {
 194       derr << "error decoding tag " << master_tag_tid << ": " << err.what()
 195            << dendl;
 196     }
 197   }
 198
 199   if (tag_data.predecessor.mirror_uuid !=
 200         librbd::Journal<>::LOCAL_MIRROR_UUID &&
 201       tag_data.predecessor.mirror_uuid !=
 202         librbd::Journal<>::ORPHAN_MIRROR_UUID) {
 203     dout(20) << "hit remote image non-primary epoch" << dendl;
 204     tag_data.predecessor.tag_tid = mirror_tag_tid;
 205   } else if (tag_data.predecessor.tag_tid == 0) {
 206     // We failed. Don't consider this fatal, just terminate retrieving.
 207     dout(20) << "making fake tag" << dendl;
 208     tag_data.predecessor.tag_tid = mirror_tag_tid;
 209   }
 210
 211   dout(20) << "decoded tag " << master_tag_tid << ": " << tag_data << dendl;
 212
 213   m_tag_cache.insert(std::make_pair(master_tag_tid, tag_data));
 214   send_update_tag_cache(tag_data.predecessor.tag_tid, mirror_tag_tid);
 215 }
 216
 217 template <typename I>
 218 void ReplayStatusFormatter<I>::format(std::string *description) {
 219
 220   dout(20) << "m_master_position=" << m_master_position
 221            << ", m_mirror_position=" << m_mirror_position
 222            << ", m_entries_behind_master=" << m_entries_behind_master << dendl;
 223
 224   std::stringstream ss;
 225   ss << "master_position=";
 226   if (m_master_position == cls::journal::ObjectPosition()) {
 227     ss << "[]";
 228   } else {
 229     ss << m_master_position;
 230   }
 231   ss << ", mirror_position=";
 232   if (m_mirror_position == cls::journal::ObjectPosition()) {
 233     ss << "[]";
 234   } else {
 235     ss << m_mirror_position;
 236   }
 237   ss << ", entries_behind_master="
 238      << (m_entries_behind_master > 0 ? m_entries_behind_master : 0);
 239
 240   *description = ss.str();
 241 }
 242
 243 } // namespace image_replayer
 244 } // namespace mirror
 245 } // namespace rbd
 246
 247 template class
 248 rbd::mirror::image_replayer::ReplayStatusFormatter<librbd::ImageCtx>;