ceph/src/tools/rbd_mirror/ImageSync.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 "ImageSync.h"
   5 #include "InstanceWatcher.h"
   6 #include "ProgressContext.h"
   7 #include "common/debug.h"
   8 #include "common/Timer.h"
   9 #include "common/errno.h"
  10 #include "journal/Journaler.h"
  11 #include "librbd/DeepCopyRequest.h"
  12 #include "librbd/ImageCtx.h"
  13 #include "librbd/ImageState.h"
  14 #include "librbd/Utils.h"
  15 #include "librbd/internal.h"
  16 #include "librbd/journal/Types.h"
  17 #include "tools/rbd_mirror/image_sync/SyncPointCreateRequest.h"
  18 #include "tools/rbd_mirror/image_sync/SyncPointPruneRequest.h"
  19
  20 #define dout_context g_ceph_context
  21 #define dout_subsys ceph_subsys_rbd_mirror
  22 #undef dout_prefix
  23 #define dout_prefix *_dout << "rbd::mirror::ImageSync: " \
  24                            << this << " " << __func__
  25
  26 namespace rbd {
  27 namespace mirror {
  28
  29 using namespace image_sync;
  30 using librbd::util::create_async_context_callback;
  31 using librbd::util::create_context_callback;
  32 using librbd::util::unique_lock_name;
  33
  34 template <typename I>
  35 class ImageSync<I>::ImageCopyProgressContext : public librbd::ProgressContext {
  36 public:
  37   ImageCopyProgressContext(ImageSync *image_sync) : image_sync(image_sync) {
  38   }
  39
  40   int update_progress(uint64_t object_no, uint64_t object_count) override {
  41     image_sync->handle_copy_image_update_progress(object_no, object_count);
  42     return 0;
  43   }
  44
  45   ImageSync *image_sync;
  46 };
  47
  48 template <typename I>
  49 ImageSync<I>::ImageSync(I *local_image_ctx, I *remote_image_ctx,
  50                         SafeTimer *timer, Mutex *timer_lock,
  51                         const std::string &mirror_uuid, Journaler *journaler,
  52                         MirrorPeerClientMeta *client_meta,
  53                         ContextWQ *work_queue,
  54                         InstanceWatcher<I> *instance_watcher,
  55                         Context *on_finish, ProgressContext *progress_ctx)
  56   : BaseRequest("rbd::mirror::ImageSync", local_image_ctx->cct, on_finish),
  57     m_local_image_ctx(local_image_ctx), m_remote_image_ctx(remote_image_ctx),
  58     m_timer(timer), m_timer_lock(timer_lock), m_mirror_uuid(mirror_uuid),
  59     m_journaler(journaler), m_client_meta(client_meta),
  60     m_work_queue(work_queue), m_instance_watcher(instance_watcher),
  61     m_progress_ctx(progress_ctx),
  62     m_lock(unique_lock_name("ImageSync::m_lock", this)),
  63     m_update_sync_point_interval(m_local_image_ctx->cct->_conf.template get_val<double>(
  64         "rbd_mirror_sync_point_update_age")), m_client_meta_copy(*client_meta) {
  65 }
  66
  67 template <typename I>
  68 ImageSync<I>::~ImageSync() {
  69   ceph_assert(m_image_copy_request == nullptr);
  70   ceph_assert(m_image_copy_prog_ctx == nullptr);
  71   ceph_assert(m_update_sync_ctx == nullptr);
  72 }
  73
  74 template <typename I>
  75 void ImageSync<I>::send() {
  76   send_notify_sync_request();
  77 }
  78
  79 template <typename I>
  80 void ImageSync<I>::cancel() {
  81   Mutex::Locker locker(m_lock);
  82
  83   dout(10) << dendl;
  84
  85   m_canceled = true;
  86
  87   if (m_instance_watcher->cancel_sync_request(m_local_image_ctx->id)) {
  88     return;
  89   }
  90
  91   if (m_image_copy_request != nullptr) {
  92     m_image_copy_request->cancel();
  93   }
  94 }
  95
  96 template <typename I>
  97 void ImageSync<I>::send_notify_sync_request() {
  98   update_progress("NOTIFY_SYNC_REQUEST");
  99
 100   dout(10) << dendl;
 101
 102   m_lock.Lock();
 103   if (m_canceled) {
 104     m_lock.Unlock();
 105     BaseRequest::finish(-ECANCELED);
 106     return;
 107   }
 108
 109   Context *ctx = create_async_context_callback(
 110     m_work_queue, create_context_callback<
 111       ImageSync<I>, &ImageSync<I>::handle_notify_sync_request>(this));
 112   m_instance_watcher->notify_sync_request(m_local_image_ctx->id, ctx);
 113   m_lock.Unlock();
 114 }
 115
 116 template <typename I>
 117 void ImageSync<I>::handle_notify_sync_request(int r) {
 118   dout(10) << ": r=" << r << dendl;
 119
 120   m_lock.Lock();
 121   if (r == 0 && m_canceled) {
 122     r = -ECANCELED;
 123   }
 124   m_lock.Unlock();
 125
 126   if (r < 0) {
 127     BaseRequest::finish(r);
 128     return;
 129   }
 130
 131   send_prune_catch_up_sync_point();
 132 }
 133
 134 template <typename I>
 135 void ImageSync<I>::send_prune_catch_up_sync_point() {
 136   update_progress("PRUNE_CATCH_UP_SYNC_POINT");
 137
 138   if (m_client_meta->sync_points.empty()) {
 139     send_create_sync_point();
 140     return;
 141   }
 142
 143   dout(10) << dendl;
 144
 145   // prune will remove sync points with missing snapshots and
 146   // ensure we have a maximum of one sync point (in case we
 147   // restarted)
 148   Context *ctx = create_context_callback<
 149     ImageSync<I>, &ImageSync<I>::handle_prune_catch_up_sync_point>(this);
 150   SyncPointPruneRequest<I> *request = SyncPointPruneRequest<I>::create(
 151     m_remote_image_ctx, false, m_journaler, m_client_meta, ctx);
 152   request->send();
 153 }
 154
 155 template <typename I>
 156 void ImageSync<I>::handle_prune_catch_up_sync_point(int r) {
 157   dout(10) << ": r=" << r << dendl;
 158
 159   if (r < 0) {
 160     derr << ": failed to prune catch-up sync point: "
 161          << cpp_strerror(r) << dendl;
 162     finish(r);
 163     return;
 164   }
 165
 166   send_create_sync_point();
 167 }
 168
 169 template <typename I>
 170 void ImageSync<I>::send_create_sync_point() {
 171   update_progress("CREATE_SYNC_POINT");
 172
 173   // TODO: when support for disconnecting laggy clients is added,
 174   //       re-connect and create catch-up sync point
 175   if (m_client_meta->sync_points.size() > 0) {
 176     send_copy_image();
 177     return;
 178   }
 179
 180   dout(10) << dendl;
 181
 182   Context *ctx = create_context_callback<
 183     ImageSync<I>, &ImageSync<I>::handle_create_sync_point>(this);
 184   SyncPointCreateRequest<I> *request = SyncPointCreateRequest<I>::create(
 185     m_remote_image_ctx, m_mirror_uuid, m_journaler, m_client_meta, ctx);
 186   request->send();
 187 }
 188
 189 template <typename I>
 190 void ImageSync<I>::handle_create_sync_point(int r) {
 191   dout(10) << ": r=" << r << dendl;
 192
 193   if (r < 0) {
 194     derr << ": failed to create sync point: " << cpp_strerror(r)
 195          << dendl;
 196     finish(r);
 197     return;
 198   }
 199
 200   send_copy_image();
 201 }
 202
 203 template <typename I>
 204 void ImageSync<I>::send_copy_image() {
 205   librados::snap_t snap_id_start = 0;
 206   librados::snap_t snap_id_end;
 207   librbd::deep_copy::ObjectNumber object_number;
 208   int r = 0;
 209   {
 210     RWLock::RLocker snap_locker(m_remote_image_ctx->snap_lock);
 211     ceph_assert(!m_client_meta->sync_points.empty());
 212     auto &sync_point = m_client_meta->sync_points.front();
 213     snap_id_end = m_remote_image_ctx->get_snap_id(
 214         cls::rbd::UserSnapshotNamespace(), sync_point.snap_name);
 215     if (snap_id_end == CEPH_NOSNAP) {
 216       derr << ": failed to locate snapshot: " << sync_point.snap_name << dendl;
 217       r = -ENOENT;
 218     } else if (!sync_point.from_snap_name.empty()) {
 219       snap_id_start = m_remote_image_ctx->get_snap_id(
 220         cls::rbd::UserSnapshotNamespace(), sync_point.from_snap_name);
 221       if (snap_id_start == CEPH_NOSNAP) {
 222         derr << ": failed to locate from snapshot: "
 223              << sync_point.from_snap_name << dendl;
 224         r = -ENOENT;
 225       }
 226     }
 227     object_number = sync_point.object_number;
 228   }
 229   if (r < 0) {
 230     finish(r);
 231     return;
 232   }
 233
 234   m_lock.Lock();
 235   if (m_canceled) {
 236     m_lock.Unlock();
 237     finish(-ECANCELED);
 238     return;
 239   }
 240
 241   dout(10) << dendl;
 242
 243   Context *ctx = create_context_callback<
 244     ImageSync<I>, &ImageSync<I>::handle_copy_image>(this);
 245   m_image_copy_prog_ctx = new ImageCopyProgressContext(this);
 246   m_image_copy_request = librbd::DeepCopyRequest<I>::create(
 247       m_remote_image_ctx, m_local_image_ctx, snap_id_start, snap_id_end,
 248       false, object_number, m_work_queue, &m_client_meta->snap_seqs,
 249       m_image_copy_prog_ctx, ctx);
 250   m_image_copy_request->get();
 251   m_lock.Unlock();
 252
 253   update_progress("COPY_IMAGE");
 254
 255   m_image_copy_request->send();
 256 }
 257
 258 template <typename I>
 259 void ImageSync<I>::handle_copy_image(int r) {
 260   dout(10) << ": r=" << r << dendl;
 261
 262   {
 263     Mutex::Locker timer_locker(*m_timer_lock);
 264     Mutex::Locker locker(m_lock);
 265     m_image_copy_request->put();
 266     m_image_copy_request = nullptr;
 267     delete m_image_copy_prog_ctx;
 268     m_image_copy_prog_ctx = nullptr;
 269     if (r == 0 && m_canceled) {
 270       r = -ECANCELED;
 271     }
 272
 273     if (m_update_sync_ctx != nullptr) {
 274       m_timer->cancel_event(m_update_sync_ctx);
 275       m_update_sync_ctx = nullptr;
 276     }
 277
 278     if (m_updating_sync_point) {
 279       m_ret_val = r;
 280       return;
 281     }
 282   }
 283
 284   if (r == -ECANCELED) {
 285     dout(10) << ": image copy canceled" << dendl;
 286     finish(r);
 287     return;
 288   } else if (r < 0) {
 289     derr << ": failed to copy image: " << cpp_strerror(r) << dendl;
 290     finish(r);
 291     return;
 292   }
 293
 294   send_flush_sync_point();
 295 }
 296
 297 template <typename I>
 298 void ImageSync<I>::handle_copy_image_update_progress(uint64_t object_no,
 299                                                      uint64_t object_count) {
 300   int percent = 100 * object_no / object_count;
 301   update_progress("COPY_IMAGE " + stringify(percent) + "%");
 302
 303   Mutex::Locker locker(m_lock);
 304   m_image_copy_object_no = object_no;
 305   m_image_copy_object_count = object_count;
 306
 307   if (m_update_sync_ctx == nullptr && !m_updating_sync_point) {
 308     send_update_sync_point();
 309   }
 310 }
 311
 312 template <typename I>
 313 void ImageSync<I>::send_update_sync_point() {
 314   ceph_assert(m_lock.is_locked());
 315
 316   m_update_sync_ctx = nullptr;
 317
 318   if (m_canceled) {
 319     return;
 320   }
 321
 322   auto sync_point = &m_client_meta->sync_points.front();
 323
 324   if (m_client_meta->sync_object_count == m_image_copy_object_count &&
 325       sync_point->object_number &&
 326       (m_image_copy_object_no - 1) == sync_point->object_number.get()) {
 327     // update sync point did not progress since last sync
 328     return;
 329   }
 330
 331   m_updating_sync_point = true;
 332
 333   m_client_meta_copy = *m_client_meta;
 334   m_client_meta->sync_object_count = m_image_copy_object_count;
 335   if (m_image_copy_object_no > 0) {
 336     sync_point->object_number = m_image_copy_object_no - 1;
 337   }
 338
 339   CephContext *cct = m_local_image_ctx->cct;
 340   ldout(cct, 20) << ": sync_point=" << *sync_point << dendl;
 341
 342   bufferlist client_data_bl;
 343   librbd::journal::ClientData client_data(*m_client_meta);
 344   encode(client_data, client_data_bl);
 345
 346   Context *ctx = create_context_callback<
 347     ImageSync<I>, &ImageSync<I>::handle_update_sync_point>(
 348       this);
 349   m_journaler->update_client(client_data_bl, ctx);
 350 }
 351
 352 template <typename I>
 353 void ImageSync<I>::handle_update_sync_point(int r) {
 354   CephContext *cct = m_local_image_ctx->cct;
 355   ldout(cct, 20) << ": r=" << r << dendl;
 356
 357   if (r < 0) {
 358     *m_client_meta = m_client_meta_copy;
 359     lderr(cct) << ": failed to update client data: " << cpp_strerror(r)
 360                << dendl;
 361   }
 362
 363   {
 364     Mutex::Locker timer_locker(*m_timer_lock);
 365     Mutex::Locker locker(m_lock);
 366     m_updating_sync_point = false;
 367
 368     if (m_image_copy_request != nullptr) {
 369       m_update_sync_ctx = new FunctionContext(
 370         [this](int r) {
 371           Mutex::Locker locker(m_lock);
 372           this->send_update_sync_point();
 373         });
 374       m_timer->add_event_after(m_update_sync_point_interval,
 375                                m_update_sync_ctx);
 376       return;
 377     }
 378   }
 379
 380   send_flush_sync_point();
 381 }
 382
 383 template <typename I>
 384 void ImageSync<I>::send_flush_sync_point() {
 385   if (m_ret_val < 0) {
 386     finish(m_ret_val);
 387     return;
 388   }
 389
 390   update_progress("FLUSH_SYNC_POINT");
 391
 392   m_client_meta_copy = *m_client_meta;
 393   m_client_meta->sync_object_count = m_image_copy_object_count;
 394   auto sync_point = &m_client_meta->sync_points.front();
 395   if (m_image_copy_object_no > 0) {
 396     sync_point->object_number = m_image_copy_object_no - 1;
 397   } else {
 398     sync_point->object_number = boost::none;
 399   }
 400
 401   dout(10) << ": sync_point=" << *sync_point << dendl;
 402
 403   bufferlist client_data_bl;
 404   librbd::journal::ClientData client_data(*m_client_meta);
 405   encode(client_data, client_data_bl);
 406
 407   Context *ctx = create_context_callback<
 408     ImageSync<I>, &ImageSync<I>::handle_flush_sync_point>(
 409       this);
 410   m_journaler->update_client(client_data_bl, ctx);
 411 }
 412
 413 template <typename I>
 414 void ImageSync<I>::handle_flush_sync_point(int r) {
 415   dout(10) << ": r=" << r << dendl;
 416
 417   if (r < 0) {
 418     *m_client_meta = m_client_meta_copy;
 419
 420     derr << ": failed to update client data: " << cpp_strerror(r)
 421          << dendl;
 422     finish(r);
 423     return;
 424   }
 425
 426   send_prune_sync_points();
 427 }
 428
 429 template <typename I>
 430 void ImageSync<I>::send_prune_sync_points() {
 431   dout(10) << dendl;
 432
 433   update_progress("PRUNE_SYNC_POINTS");
 434
 435   Context *ctx = create_context_callback<
 436     ImageSync<I>, &ImageSync<I>::handle_prune_sync_points>(this);
 437   SyncPointPruneRequest<I> *request = SyncPointPruneRequest<I>::create(
 438     m_remote_image_ctx, true, m_journaler, m_client_meta, ctx);
 439   request->send();
 440 }
 441
 442 template <typename I>
 443 void ImageSync<I>::handle_prune_sync_points(int r) {
 444   dout(10) << ": r=" << r << dendl;
 445
 446   if (r < 0) {
 447     derr << ": failed to prune sync point: "
 448          << cpp_strerror(r) << dendl;
 449     finish(r);
 450     return;
 451   }
 452
 453   if (!m_client_meta->sync_points.empty()) {
 454     send_copy_image();
 455     return;
 456   }
 457
 458   finish(0);
 459 }
 460
 461 template <typename I>
 462 void ImageSync<I>::update_progress(const std::string &description) {
 463   dout(20) << ": " << description << dendl;
 464
 465   if (m_progress_ctx) {
 466     m_progress_ctx->update_progress("IMAGE_SYNC/" + description);
 467   }
 468 }
 469
 470 template <typename I>
 471 void ImageSync<I>::finish(int r) {
 472   dout(20) << ": r=" << r << dendl;
 473
 474   m_instance_watcher->notify_sync_complete(m_local_image_ctx->id);
 475   BaseRequest::finish(r);
 476 }
 477
 478 } // namespace mirror
 479 } // namespace rbd
 480
 481 template class rbd::mirror::ImageSync<librbd::ImageCtx>;