Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

45 files changed, 6505 insertions, 0 deletions

diff --git a/src/librbd/mirror/DemoteRequest.cc b/src/librbd/mirror/DemoteRequest.cc
new file mode 100644
index 000000000..350a76d83
--- /dev/null
+++ b/src/librbd/mirror/DemoteRequest.cc
@@ -0,0 +1,216 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/DemoteRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ExclusiveLock.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/GetInfoRequest.h"
+#include "librbd/mirror/snapshot/DemoteRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::DemoteRequest: " << this \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using librbd::util::create_context_callback;
+
+template <typename I>
+void DemoteRequest<I>::send() {
+  get_info();
+}
+
+template <typename I>
+void DemoteRequest<I>::get_info() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    DemoteRequest<I>, &DemoteRequest<I>::handle_get_info>(this);
+  auto req = GetInfoRequest<I>::create(m_image_ctx, &m_mirror_image,
+                                       &m_promotion_state,
+                                       &m_primary_mirror_uuid, ctx);
+  req->send();
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_get_info(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  } else if (m_mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+    lderr(cct) << "mirroring is not currently enabled" << dendl;
+    finish(-EINVAL);
+    return;
+  } else if (m_promotion_state != PROMOTION_STATE_PRIMARY) {
+    lderr(cct) << "image is not primary" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  acquire_lock();
+}
+
+template <typename I>
+void DemoteRequest<I>::acquire_lock() {
+  CephContext *cct = m_image_ctx.cct;
+
+  m_image_ctx.owner_lock.lock_shared();
+  if (m_image_ctx.exclusive_lock == nullptr) {
+    m_image_ctx.owner_lock.unlock_shared();
+    if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
+      lderr(cct) << "exclusive lock is not active" << dendl;
+      finish(-EINVAL);
+    } else {
+      demote();
+    }
+    return;
+  }
+
+  // avoid accepting new requests from peers while we demote
+  // the image
+  m_image_ctx.exclusive_lock->block_requests(0);
+  m_blocked_requests = true;
+
+  if (m_image_ctx.exclusive_lock->is_lock_owner()) {
+    m_image_ctx.owner_lock.unlock_shared();
+    demote();
+    return;
+  }
+
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    DemoteRequest<I>,
+    &DemoteRequest<I>::handle_acquire_lock>(this, m_image_ctx.exclusive_lock);
+  m_image_ctx.exclusive_lock->acquire_lock(ctx);
+  m_image_ctx.owner_lock.unlock_shared();
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_acquire_lock(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to lock image: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  m_image_ctx.owner_lock.lock_shared();
+  if (m_image_ctx.exclusive_lock != nullptr &&
+      !m_image_ctx.exclusive_lock->is_lock_owner()) {
+    r = m_image_ctx.exclusive_lock->get_unlocked_op_error();
+    m_image_ctx.owner_lock.unlock_shared();
+    lderr(cct) << "failed to acquire exclusive lock" << dendl;
+    finish(r);
+    return;
+  }
+  m_image_ctx.owner_lock.unlock_shared();
+
+  demote();
+}
+
+template <typename I>
+void DemoteRequest<I>::demote() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    DemoteRequest<I>, &DemoteRequest<I>::handle_demote>(this);
+  if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
+    Journal<I>::demote(&m_image_ctx, ctx);
+  } else if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    auto req = mirror::snapshot::DemoteRequest<I>::create(
+      &m_image_ctx, m_mirror_image.global_image_id, ctx);
+    req->send();
+  } else {
+    lderr(cct) << "unknown image mirror mode: " << m_mirror_image.mode << dendl;
+    m_ret_val = -EOPNOTSUPP;
+    release_lock();
+  }
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_demote(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    m_ret_val = r;
+    lderr(cct) << "failed to demote image: " << cpp_strerror(r) << dendl;
+  }
+
+  release_lock();
+}
+
+template <typename I>
+void DemoteRequest<I>::release_lock() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  m_image_ctx.owner_lock.lock_shared();
+  if (m_image_ctx.exclusive_lock == nullptr) {
+    m_image_ctx.owner_lock.unlock_shared();
+    finish(0);
+    return;
+  }
+
+  auto ctx = create_context_callback<
+    DemoteRequest<I>,
+    &DemoteRequest<I>::handle_release_lock>(this, m_image_ctx.exclusive_lock);
+  m_image_ctx.exclusive_lock->release_lock(ctx);
+  m_image_ctx.owner_lock.unlock_shared();
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_release_lock(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to release exclusive lock: " << cpp_strerror(r)
+               << dendl;
+  }
+
+  finish(r);
+}
+
+template <typename I>
+void DemoteRequest<I>::finish(int r) {
+  if (m_ret_val < 0) {
+    r = m_ret_val;
+  }
+
+  {
+    std::shared_lock owner_locker{m_image_ctx.owner_lock};
+    if (m_blocked_requests && m_image_ctx.exclusive_lock != nullptr) {
+      m_image_ctx.exclusive_lock->unblock_requests();
+    }
+  }
+
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::DemoteRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/DemoteRequest.h b/src/librbd/mirror/DemoteRequest.h
new file mode 100644
index 000000000..ab9239068
--- /dev/null
+++ b/src/librbd/mirror/DemoteRequest.h
@@ -0,0 +1,86 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_DEMOTE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_DEMOTE_REQUEST_H
+
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/mirror/Types.h"
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class DemoteRequest {
+public:
+  static DemoteRequest *create(ImageCtxT &image_ctx, Context *on_finish) {
+    return new DemoteRequest(image_ctx, on_finish);
+  }
+
+  DemoteRequest(ImageCtxT &image_ctx, Context *on_finish)
+    : m_image_ctx(image_ctx), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_INFO
+   *    |
+   *    v
+   * ACQUIRE_LOCK * * * *
+   *    |               *
+   *    v               *
+   * DEMOTE             *
+   *    |               *
+   *    v               *
+   * RELEASE_LOCK       *
+   *    |               *
+   *    v               *
+   * <finish> < * * * * *
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT &m_image_ctx;
+  Context *m_on_finish;
+
+  int m_ret_val = 0;
+  bool m_blocked_requests = false;
+
+  cls::rbd::MirrorImage m_mirror_image;
+  PromotionState m_promotion_state = PROMOTION_STATE_PRIMARY;
+  std::string m_primary_mirror_uuid;
+
+  void get_info();
+  void handle_get_info(int r);
+
+  void acquire_lock();
+  void handle_acquire_lock(int r);
+
+  void demote();
+  void handle_demote(int r);
+
+  void release_lock();
+  void handle_release_lock(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::DemoteRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_DEMOTE_REQUEST_H
diff --git a/src/librbd/mirror/DisableRequest.cc b/src/librbd/mirror/DisableRequest.cc
new file mode 100644
index 000000000..09378ce58
--- /dev/null
+++ b/src/librbd/mirror/DisableRequest.cc
@@ -0,0 +1,479 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/DisableRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/journal/cls_journal_client.h"
+#include "journal/Journaler.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+#include "librbd/asio/ContextWQ.h"
+#include "librbd/journal/PromoteRequest.h"
+#include "librbd/mirror/GetInfoRequest.h"
+#include "librbd/mirror/ImageRemoveRequest.h"
+#include "librbd/mirror/ImageStateUpdateRequest.h"
+#include "librbd/mirror/snapshot/PromoteRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::DisableRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using util::create_rados_callback;
+
+template <typename I>
+DisableRequest<I>::DisableRequest(I *image_ctx, bool force, bool remove,
+                                  Context *on_finish)
+  : m_image_ctx(image_ctx), m_force(force), m_remove(remove),
+    m_on_finish(on_finish) {
+}
+
+template <typename I>
+void DisableRequest<I>::send() {
+  send_get_mirror_info();
+}
+
+template <typename I>
+void DisableRequest<I>::send_get_mirror_info() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+
+  using klass = DisableRequest<I>;
+  Context *ctx = util::create_context_callback<
+      klass, &klass::handle_get_mirror_info>(this);
+
+  auto req = GetInfoRequest<I>::create(*m_image_ctx, &m_mirror_image,
+                                       &m_promotion_state,
+                                       &m_primary_mirror_uuid, ctx);
+  req->send();
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_get_mirror_info(int *result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  if (*result < 0) {
+    if (*result == -ENOENT) {
+      ldout(cct, 20) << "mirroring is not enabled for this image" << dendl;
+      *result = 0;
+    } else {
+      lderr(cct) << "failed to get mirroring info: " << cpp_strerror(*result)
+                 << dendl;
+    }
+    return m_on_finish;
+  }
+
+  m_is_primary = (m_promotion_state == PROMOTION_STATE_PRIMARY ||
+                  m_promotion_state == PROMOTION_STATE_UNKNOWN);
+
+  if (!m_is_primary && !m_force) {
+    lderr(cct) << "mirrored image is not primary, "
+               << "add force option to disable mirroring" << dendl;
+    *result = -EINVAL;
+    return m_on_finish;
+  }
+
+  send_image_state_update();
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::send_image_state_update() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    DisableRequest<I>,
+    &DisableRequest<I>::handle_image_state_update>(this);
+  auto req = ImageStateUpdateRequest<I>::create(
+    m_image_ctx->md_ctx, m_image_ctx->id,
+    cls::rbd::MIRROR_IMAGE_STATE_DISABLING, m_mirror_image, ctx);
+  req->send();
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_image_state_update(int *result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  if (*result < 0) {
+    lderr(cct) << "failed to disable mirroring: " << cpp_strerror(*result)
+               << dendl;
+    return m_on_finish;
+  }
+
+  send_promote_image();
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::send_promote_image() {
+  if (m_is_primary) {
+    clean_mirror_state();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    DisableRequest<I>, &DisableRequest<I>::handle_promote_image>(this);
+  if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
+    // Not primary -- shouldn't have the journal open
+    ceph_assert(m_image_ctx->journal == nullptr);
+
+    auto req = journal::PromoteRequest<I>::create(m_image_ctx, true, ctx);
+    req->send();
+  } else if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    auto req = mirror::snapshot::PromoteRequest<I>::create(
+      m_image_ctx, m_mirror_image.global_image_id, ctx);
+    req->send();
+  } else {
+    lderr(cct) << "unknown image mirror mode: " << m_mirror_image.mode << dendl;
+    ctx->complete(-EOPNOTSUPP);
+  }
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_promote_image(int *result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  if (*result < 0) {
+    lderr(cct) << "failed to promote image: " << cpp_strerror(*result) << dendl;
+    return m_on_finish;
+  }
+
+  send_refresh_image();
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::send_refresh_image() {
+  if (!m_image_ctx->state->is_refresh_required()) {
+    clean_mirror_state();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    DisableRequest<I>,
+    &DisableRequest<I>::handle_refresh_image>(this);
+  m_image_ctx->state->refresh(ctx);
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_refresh_image(int* result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  if (*result < 0) {
+    lderr(cct) << "failed to refresh image: " << cpp_strerror(*result) << dendl;
+    return m_on_finish;
+  }
+
+  clean_mirror_state();
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::clean_mirror_state() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    remove_mirror_snapshots();
+  } else {
+    send_get_clients();
+  }
+}
+
+template <typename I>
+void DisableRequest<I>::send_get_clients() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  using klass = DisableRequest<I>;
+  Context *ctx = util::create_context_callback<
+    klass, &klass::handle_get_clients>(this);
+
+  std::string header_oid = ::journal::Journaler::header_oid(m_image_ctx->id);
+  m_clients.clear();
+  cls::journal::client::client_list(m_image_ctx->md_ctx, header_oid, &m_clients,
+                                    ctx);
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_get_clients(int *result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  std::unique_lock locker{m_lock};
+  ceph_assert(m_current_ops.empty());
+
+  if (*result < 0) {
+    lderr(cct) << "failed to get registered clients: " << cpp_strerror(*result)
+               << dendl;
+    return m_on_finish;
+  }
+
+  for (auto client : m_clients) {
+    journal::ClientData client_data;
+    auto bl_it = client.data.cbegin();
+    try {
+      using ceph::decode;
+      decode(client_data, bl_it);
+    } catch (const buffer::error &err) {
+      lderr(cct) << "failed to decode client data" << dendl;
+      m_error_result = -EBADMSG;
+      continue;
+    }
+
+    journal::ClientMetaType type = client_data.get_client_meta_type();
+    if (type != journal::ClientMetaType::MIRROR_PEER_CLIENT_META_TYPE) {
+      continue;
+    }
+
+    if (m_current_ops.find(client.id) != m_current_ops.end()) {
+      // Should not happen.
+      lderr(cct) << "clients with the same id "
+                 << client.id << dendl;
+      continue;
+    }
+
+    m_current_ops[client.id] = 0;
+    m_ret[client.id] = 0;
+
+    journal::MirrorPeerClientMeta client_meta =
+      boost::get<journal::MirrorPeerClientMeta>(client_data.client_meta);
+
+    for (const auto& sync : client_meta.sync_points) {
+      send_remove_snap(client.id, sync.snap_namespace, sync.snap_name);
+    }
+
+    if (m_current_ops[client.id] == 0) {
+      // no snaps to remove
+      send_unregister_client(client.id);
+    }
+  }
+
+  if (m_current_ops.empty()) {
+    if (m_error_result < 0) {
+      *result = m_error_result;
+      return m_on_finish;
+    } else if (!m_remove) {
+      return m_on_finish;
+    }
+    locker.unlock();
+
+    // no mirror clients to unregister
+    send_remove_mirror_image();
+  }
+
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::remove_mirror_snapshots() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  // remove snapshot-based mirroring snapshots
+  bool removing_snapshots = false;
+  {
+    std::lock_guard locker{m_lock};
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+
+    for (auto &it : m_image_ctx->snap_info) {
+      auto &snap_info = it.second;
+      auto type = cls::rbd::get_snap_namespace_type(
+        snap_info.snap_namespace);
+      if (type == cls::rbd::SNAPSHOT_NAMESPACE_TYPE_MIRROR) {
+        send_remove_snap("", snap_info.snap_namespace, snap_info.name);
+        removing_snapshots = true;
+      }
+    }
+  }
+
+  if (!removing_snapshots) {
+    send_remove_mirror_image();
+  }
+}
+
+template <typename I>
+void DisableRequest<I>::send_remove_snap(
+    const std::string &client_id,
+    const cls::rbd::SnapshotNamespace &snap_namespace,
+    const std::string &snap_name) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "client_id=" << client_id
+                 << ", snap_name=" << snap_name << dendl;
+
+  ceph_assert(ceph_mutex_is_locked(m_lock));
+
+  m_current_ops[client_id]++;
+
+  Context *ctx = create_context_callback(
+    &DisableRequest<I>::handle_remove_snap, client_id);
+
+  ctx = new LambdaContext([this, snap_namespace, snap_name, ctx](int r) {
+      m_image_ctx->operations->snap_remove(snap_namespace,
+                                           snap_name.c_str(),
+                                           ctx);
+    });
+
+  m_image_ctx->op_work_queue->queue(ctx, 0);
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_remove_snap(int *result,
+    const std::string &client_id) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  std::unique_lock locker{m_lock};
+
+  ceph_assert(m_current_ops[client_id] > 0);
+  m_current_ops[client_id]--;
+
+  if (*result < 0 && *result != -ENOENT) {
+    lderr(cct) << "failed to remove mirroring snapshot: "
+               << cpp_strerror(*result) << dendl;
+    m_ret[client_id] = *result;
+  }
+
+  if (m_current_ops[client_id] == 0) {
+    if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+      ceph_assert(client_id.empty());
+      m_current_ops.erase(client_id);
+      if (m_ret[client_id] < 0) {
+        return m_on_finish;
+      }
+      locker.unlock();
+
+      send_remove_mirror_image();
+      return nullptr;
+    }
+
+    send_unregister_client(client_id);
+  }
+
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::send_unregister_client(
+  const std::string &client_id) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  ceph_assert(ceph_mutex_is_locked(m_lock));
+  ceph_assert(m_current_ops[client_id] == 0);
+
+  Context *ctx = create_context_callback(
+    &DisableRequest<I>::handle_unregister_client, client_id);
+
+  if (m_ret[client_id] < 0) {
+    m_image_ctx->op_work_queue->queue(ctx, m_ret[client_id]);
+    return;
+  }
+
+  librados::ObjectWriteOperation op;
+  cls::journal::client::client_unregister(&op, client_id);
+  std::string header_oid = ::journal::Journaler::header_oid(m_image_ctx->id);
+  librados::AioCompletion *comp = create_rados_callback(ctx);
+
+  int r = m_image_ctx->md_ctx.aio_operate(header_oid, comp, &op);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_unregister_client(
+  int *result, const std::string &client_id) {
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  std::unique_lock locker{m_lock};
+  ceph_assert(m_current_ops[client_id] == 0);
+  m_current_ops.erase(client_id);
+
+  if (*result < 0 && *result != -ENOENT) {
+    lderr(cct) << "failed to unregister remote journal client: "
+               << cpp_strerror(*result) << dendl;
+    m_error_result = *result;
+  }
+
+  if (!m_current_ops.empty()) {
+    return nullptr;
+  }
+
+  if (m_error_result < 0) {
+    *result = m_error_result;
+    return m_on_finish;
+  }
+  locker.unlock();
+
+  send_get_clients();
+  return nullptr;
+}
+
+template <typename I>
+void DisableRequest<I>::send_remove_mirror_image() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    DisableRequest<I>,
+    &DisableRequest<I>::handle_remove_mirror_image>(this);
+  auto req = ImageRemoveRequest<I>::create(
+    m_image_ctx->md_ctx, m_mirror_image.global_image_id, m_image_ctx->id,
+    ctx);
+  req->send();
+}
+
+template <typename I>
+Context *DisableRequest<I>::handle_remove_mirror_image(int *result) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << *result << dendl;
+
+  if (*result < 0) {
+    lderr(cct) << "failed to remove mirror image: " << cpp_strerror(*result)
+               << dendl;
+    return m_on_finish;
+  }
+
+  ldout(cct, 20) << "removed image state from rbd_mirroring object" << dendl;
+  return m_on_finish;
+}
+
+template <typename I>
+Context *DisableRequest<I>::create_context_callback(
+  Context*(DisableRequest<I>::*handle)(int*, const std::string &client_id),
+  const std::string &client_id) {
+
+  return new LambdaContext([this, handle, client_id](int r) {
+      Context *on_finish = (this->*handle)(&r, client_id);
+      if (on_finish != nullptr) {
+        on_finish->complete(r);
+        delete this;
+      }
+    });
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::DisableRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/DisableRequest.h b/src/librbd/mirror/DisableRequest.h
new file mode 100644
index 000000000..f45d1a14c
--- /dev/null
+++ b/src/librbd/mirror/DisableRequest.h
@@ -0,0 +1,143 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_DISABLE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_DISABLE_REQUEST_H
+
+#include "include/buffer.h"
+#include "common/ceph_mutex.h"
+#include "cls/journal/cls_journal_types.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/mirror/Types.h"
+
+#include <map>
+#include <string>
+
+class Context;
+
+namespace librbd {
+
+class ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = ImageCtx>
+class DisableRequest {
+public:
+  static DisableRequest *create(ImageCtxT *image_ctx, bool force,
+                                bool remove, Context *on_finish) {
+    return new DisableRequest(image_ctx, force, remove, on_finish);
+  }
+
+  DisableRequest(ImageCtxT *image_ctx, bool force, bool remove,
+                 Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_MIRROR_INFO  * * * * * * * * * * * * * * * * * * * * * * *
+   *    |                                                         *
+   *    v                                                         *
+   * IMAGE_STATE_UPDATE * * * * * * * * * * * * * * * * * * * * * *
+   *    |                                                         *
+   *    v                                                         *
+   * PROMOTE_IMAGE (skip if primary)                              *
+   *    |                                                         *
+   *    v                                                         *
+   * REFRESH_IMAGE (skip if necessary)                            *
+   *    |                                                         *
+   *    v                                                         *
+   * GET_CLIENTS <----------------------------------------\ * * * *
+   *    |     | (unregister clients)                      |       *  (on error)
+   *    |     |/----------------------------\             |       *
+   *    |     |                             |             |       *
+   *    |     |   /-----------\ (repeat     | (repeat     | (repeat
+   *    |     |   |           |  as needed) |  as needed) |  as needed)
+   *    |     v   v           |             |             |       *
+   *    |  REMOVE_SYNC_SNAP --/ * * * * * * | * * * * * * | * * * *
+   *    |     |                             |             |       *
+   *    |     v                             |             |       *
+   *    |  UNREGISTER_CLIENT ---------------/-------------/ * * * *
+   *    |                                                         *
+   *    | (no more clients                                        *
+   *    |  to unregister)                                         *
+   *    v                                                         *
+   * REMOVE_MIRROR_IMAGE  * * * * * * * * * * * * * * * * * * * * *
+   *    |         (skip if no remove)                             *
+   *    v                                                         *
+   * <finish> < * * * * * * * * * * * * * * * * * * * * * * * * * *
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  bool m_force;
+  bool m_remove;
+  Context *m_on_finish;
+
+  bool m_is_primary = false;
+  cls::rbd::MirrorImage m_mirror_image;
+  PromotionState m_promotion_state = PROMOTION_STATE_NON_PRIMARY;
+  std::string m_primary_mirror_uuid;
+  std::set<cls::journal::Client> m_clients;
+  std::map<std::string, int> m_ret;
+  std::map<std::string, int> m_current_ops;
+  int m_error_result = 0;
+  mutable ceph::mutex m_lock =
+    ceph::make_mutex("mirror::DisableRequest::m_lock");
+
+  void send_get_mirror_info();
+  Context *handle_get_mirror_info(int *result);
+
+  void send_image_state_update();
+  Context *handle_image_state_update(int *result);
+
+  void send_notify_mirroring_watcher();
+  Context *handle_notify_mirroring_watcher(int *result);
+
+  void send_promote_image();
+  Context *handle_promote_image(int *result);
+
+  void send_refresh_image();
+  Context* handle_refresh_image(int* result);
+
+  void clean_mirror_state();
+
+  void send_get_clients();
+  Context *handle_get_clients(int *result);
+
+  void remove_mirror_snapshots();
+
+  void send_remove_snap(const std::string &client_id,
+                        const cls::rbd::SnapshotNamespace &snap_namespace,
+			const std::string &snap_name);
+  Context *handle_remove_snap(int *result, const std::string &client_id);
+
+  void send_unregister_client(const std::string &client_id);
+  Context *handle_unregister_client(int *result, const std::string &client_id);
+
+  void send_remove_mirror_image();
+  Context *handle_remove_mirror_image(int *result);
+
+  void send_notify_mirroring_watcher_removed();
+  Context *handle_notify_mirroring_watcher_removed(int *result);
+
+  Context *create_context_callback(
+    Context*(DisableRequest<ImageCtxT>::*handle)(
+      int*, const std::string &client_id),
+    const std::string &client_id);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::DisableRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_DISABLE_REQUEST_H
diff --git a/src/librbd/mirror/EnableRequest.cc b/src/librbd/mirror/EnableRequest.cc
new file mode 100644
index 000000000..fd74a25ba
--- /dev/null
+++ b/src/librbd/mirror/EnableRequest.cc
@@ -0,0 +1,329 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/EnableRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/ImageStateUpdateRequest.h"
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::EnableRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using util::create_context_callback;
+using util::create_rados_callback;
+
+template <typename I>
+EnableRequest<I>::EnableRequest(librados::IoCtx &io_ctx,
+                                const std::string &image_id,
+                                I* image_ctx,
+                                cls::rbd::MirrorImageMode mode,
+                                const std::string &non_primary_global_image_id,
+                                bool image_clean,
+                                asio::ContextWQ *op_work_queue,
+                                Context *on_finish)
+  : m_io_ctx(io_ctx), m_image_id(image_id), m_image_ctx(image_ctx),
+    m_mode(mode), m_non_primary_global_image_id(non_primary_global_image_id),
+    m_image_clean(image_clean), m_op_work_queue(op_work_queue),
+    m_on_finish(on_finish),
+    m_cct(reinterpret_cast<CephContext*>(io_ctx.cct())) {
+}
+
+template <typename I>
+void EnableRequest<I>::send() {
+  get_mirror_image();
+}
+
+template <typename I>
+void EnableRequest<I>::get_mirror_image() {
+  ldout(m_cct, 10) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_get_start(&op, m_image_id);
+
+  using klass = EnableRequest<I>;
+  librados::AioCompletion *comp =
+    create_rados_callback<klass, &klass::handle_get_mirror_image>(this);
+  m_out_bl.clear();
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void EnableRequest<I>::handle_get_mirror_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_image_get_finish(&iter, &m_mirror_image);
+  }
+
+  if (r == 0 && m_mirror_image.state == cls::rbd::MIRROR_IMAGE_STATE_CREATING &&
+      !m_non_primary_global_image_id.empty()) {
+    // special case where rbd-mirror injects a disabled record to record the
+    // local image id prior to creating ther image
+    ldout(m_cct, 10) << "enabling mirroring on in-progress image replication"
+                     << dendl;
+  } else if (r == 0) {
+    if (m_mirror_image.mode != m_mode) {
+      lderr(m_cct) << "invalid current image mirror mode" << dendl;
+      r = -EINVAL;
+    } else if (m_mirror_image.state == cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+      ldout(m_cct, 10) << "mirroring is already enabled" << dendl;
+    } else {
+      lderr(m_cct) << "currently disabling" << dendl;
+      r = -EINVAL;
+    }
+    finish(r);
+    return;
+  } else if (r != -ENOENT) {
+    lderr(m_cct) << "failed to retrieve mirror image: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  r = 0;
+  m_mirror_image.mode = m_mode;
+  if (m_non_primary_global_image_id.empty()) {
+    uuid_d uuid_gen;
+    uuid_gen.generate_random();
+    m_mirror_image.global_image_id = uuid_gen.to_string();
+  } else {
+    m_mirror_image.global_image_id = m_non_primary_global_image_id;
+  }
+
+  get_tag_owner();
+}
+
+template <typename I>
+void EnableRequest<I>::get_tag_owner() {
+  if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    open_image();
+    return;
+  } else if (!m_non_primary_global_image_id.empty()) {
+    image_state_update();
+    return;
+  }
+
+  ldout(m_cct, 10)  << dendl;
+
+  using klass = EnableRequest<I>;
+  Context *ctx = create_context_callback<
+      klass, &klass::handle_get_tag_owner>(this);
+  librbd::Journal<>::is_tag_owner(m_io_ctx, m_image_id, &m_is_primary,
+                                  m_op_work_queue, ctx);
+}
+
+template <typename I>
+void EnableRequest<I>::handle_get_tag_owner(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to check tag ownership: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  if (!m_is_primary) {
+    lderr(m_cct) << "last journal tag not owned by local cluster" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  image_state_update();
+}
+
+template <typename I>
+void EnableRequest<I>::open_image() {
+  if (!m_non_primary_global_image_id.empty()) {
+    // special case for rbd-mirror creating a non-primary image
+    enable_non_primary_feature();
+    return;
+  } else if (m_image_ctx != nullptr) {
+    create_primary_snapshot();
+    return;
+  }
+
+  ldout(m_cct, 10) << dendl;
+
+  m_close_image = true;
+  m_image_ctx = I::create("", m_image_id, CEPH_NOSNAP, m_io_ctx, false);
+
+  auto ctx = create_context_callback<
+    EnableRequest<I>, &EnableRequest<I>::handle_open_image>(this);
+  m_image_ctx->state->open(OPEN_FLAG_SKIP_OPEN_PARENT |
+                           OPEN_FLAG_IGNORE_MIGRATING, ctx);
+}
+
+template <typename I>
+void EnableRequest<I>::handle_open_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to open image: " << cpp_strerror(r) << dendl;
+    m_image_ctx = nullptr;
+    finish(r);
+    return;
+  }
+
+  create_primary_snapshot();
+}
+
+template <typename I>
+void EnableRequest<I>::create_primary_snapshot() {
+  ldout(m_cct, 10) << dendl;
+
+  ceph_assert(m_image_ctx != nullptr);
+  uint64_t snap_create_flags;
+  int r = util::snap_create_flags_api_to_internal(
+      m_cct, util::get_default_snap_create_flags(m_image_ctx),
+      &snap_create_flags);
+  ceph_assert(r == 0);
+  auto ctx = create_context_callback<
+    EnableRequest<I>,
+    &EnableRequest<I>::handle_create_primary_snapshot>(this);
+  auto req = snapshot::CreatePrimaryRequest<I>::create(
+    m_image_ctx, m_mirror_image.global_image_id,
+    (m_image_clean ? 0 : CEPH_NOSNAP), snap_create_flags,
+    snapshot::CREATE_PRIMARY_FLAG_IGNORE_EMPTY_PEERS, &m_snap_id, ctx);
+  req->send();
+}
+
+template <typename I>
+void EnableRequest<I>::handle_create_primary_snapshot(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to create initial primary snapshot: "
+                 << cpp_strerror(r) << dendl;
+    m_ret_val = r;
+  }
+
+  close_image();
+}
+
+template <typename I>
+void EnableRequest<I>::close_image() {
+  if (!m_close_image) {
+    if (m_ret_val < 0) {
+      finish(m_ret_val);
+    } else {
+      image_state_update();
+    }
+    return;
+  }
+
+  ldout(m_cct, 10) << dendl;
+
+  auto ctx = create_context_callback<
+    EnableRequest<I>, &EnableRequest<I>::handle_close_image>(this);
+  m_image_ctx->state->close(ctx);
+}
+
+template <typename I>
+void EnableRequest<I>::handle_close_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  m_image_ctx = nullptr;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to close image: " << cpp_strerror(r) << dendl;
+    if (m_ret_val == 0) {
+      m_ret_val = r;
+    }
+  }
+
+  if (m_ret_val < 0) {
+    finish(m_ret_val);
+    return;
+  }
+
+  image_state_update();
+}
+
+
+template <typename I>
+void EnableRequest<I>::enable_non_primary_feature() {
+  if (m_mirror_image.mode != cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    image_state_update();
+    return;
+  }
+
+  ldout(m_cct, 10) << dendl;
+
+  // ensure image is flagged with non-primary feature so that
+  // standard RBD clients cannot write to it.
+  librados::ObjectWriteOperation op;
+  cls_client::set_features(&op, RBD_FEATURE_NON_PRIMARY,
+                           RBD_FEATURE_NON_PRIMARY);
+
+  auto aio_comp = create_rados_callback<
+    EnableRequest<I>,
+    &EnableRequest<I>::handle_enable_non_primary_feature>(this);
+  int r = m_io_ctx.aio_operate(util::header_name(m_image_id), aio_comp, &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void EnableRequest<I>::handle_enable_non_primary_feature(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to enable non-primary feature: "
+                 << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void EnableRequest<I>::image_state_update() {
+  ldout(m_cct, 10) << dendl;
+
+  auto ctx = create_context_callback<
+    EnableRequest<I>, &EnableRequest<I>::handle_image_state_update>(this);
+  auto req = ImageStateUpdateRequest<I>::create(
+    m_io_ctx, m_image_id, cls::rbd::MIRROR_IMAGE_STATE_ENABLED,
+    m_mirror_image, ctx);
+  req->send();
+}
+
+template <typename I>
+void EnableRequest<I>::handle_image_state_update(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to enable mirroring: " << cpp_strerror(r)
+                 << dendl;
+  }
+
+  finish(r);
+}
+
+template <typename I>
+void EnableRequest<I>::finish(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::EnableRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/EnableRequest.h b/src/librbd/mirror/EnableRequest.h
new file mode 100644
index 000000000..391028e6e
--- /dev/null
+++ b/src/librbd/mirror/EnableRequest.h
@@ -0,0 +1,135 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_ENABLE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_ENABLE_REQUEST_H
+
+#include "include/buffer_fwd.h"
+#include "include/rados/librados_fwd.hpp"
+#include "include/rbd/librbd.hpp"
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/mirror/Types.h"
+#include <map>
+#include <string>
+
+class Context;
+
+namespace librbd {
+
+namespace asio { struct ContextWQ; }
+
+namespace mirror {
+
+template <typename ImageCtxT = ImageCtx>
+class EnableRequest {
+public:
+  static EnableRequest *create(ImageCtxT *image_ctx,
+                               cls::rbd::MirrorImageMode mode,
+                               const std::string &non_primary_global_image_id,
+                               bool image_clean, Context *on_finish) {
+    return new EnableRequest(image_ctx->md_ctx, image_ctx->id, image_ctx, mode,
+                             non_primary_global_image_id, image_clean,
+                             image_ctx->op_work_queue, on_finish);
+  }
+  static EnableRequest *create(librados::IoCtx &io_ctx,
+                               const std::string &image_id,
+                               cls::rbd::MirrorImageMode mode,
+                               const std::string &non_primary_global_image_id,
+                               bool image_clean, asio::ContextWQ *op_work_queue,
+                               Context *on_finish) {
+    return new EnableRequest(io_ctx, image_id, nullptr, mode,
+                             non_primary_global_image_id, image_clean,
+                             op_work_queue, on_finish);
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_MIRROR_IMAGE * * * * * * *
+   *    |                         * (on error)
+   *    v (skip if not needed)    *
+   * GET_TAG_OWNER  * * * * * * * *
+   *    |                         *
+   *    v (skip if not needed)    *
+   * OPEN_IMAGE                   *
+   *    |                         *
+   *    v (skip if not needed)    *
+   * CREATE_PRIMARY_SNAPSHOT  * * *
+   *    |                         *
+   *    v (skip of not opened)    *
+   * CLOSE_IMAGE                  *
+   *    |                         *
+   *    v (skip if not needed)    *
+   * ENABLE_NON_PRIMARY_FEATURE   *
+   *    |                         *
+   *    v (skip if not needed)    *
+   * IMAGE_STATE_UPDATE * * * * * *
+   *    |                         *
+   *    v                         *
+   * <finish>   < * * * * * * * * *
+   *
+   * @endverbatim
+   */
+
+  EnableRequest(librados::IoCtx &io_ctx, const std::string &image_id,
+                ImageCtxT* image_ctx, cls::rbd::MirrorImageMode mode,
+                const std::string &non_primary_global_image_id,
+                bool image_clean, asio::ContextWQ *op_work_queue,
+                Context *on_finish);
+
+  librados::IoCtx &m_io_ctx;
+  std::string m_image_id;
+  ImageCtxT* m_image_ctx;
+  cls::rbd::MirrorImageMode m_mode;
+  std::string m_non_primary_global_image_id;
+  bool m_image_clean;
+  asio::ContextWQ *m_op_work_queue;
+  Context *m_on_finish;
+
+  CephContext *m_cct = nullptr;
+  bufferlist m_out_bl;
+  cls::rbd::MirrorImage m_mirror_image;
+
+  int m_ret_val = 0;
+  bool m_close_image = false;
+
+  bool m_is_primary = false;
+  uint64_t m_snap_id = CEPH_NOSNAP;
+
+  void get_mirror_image();
+  void handle_get_mirror_image(int r);
+
+  void get_tag_owner();
+  void handle_get_tag_owner(int r);
+
+  void open_image();
+  void handle_open_image(int r);
+
+  void create_primary_snapshot();
+  void handle_create_primary_snapshot(int r);
+
+  void close_image();
+  void handle_close_image(int r);
+
+  void enable_non_primary_feature();
+  void handle_enable_non_primary_feature(int r);
+
+  void image_state_update();
+  void handle_image_state_update(int r);
+
+  void finish(int r);
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::EnableRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_ENABLE_REQUEST_H
diff --git a/src/librbd/mirror/GetInfoRequest.cc b/src/librbd/mirror/GetInfoRequest.cc
new file mode 100644
index 000000000..a7ee64567
--- /dev/null
+++ b/src/librbd/mirror/GetInfoRequest.cc
@@ -0,0 +1,290 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/GetInfoRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::GetInfoRequest: " << this \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+GetInfoRequest<I>::GetInfoRequest(librados::IoCtx& io_ctx,
+                                  asio::ContextWQ *op_work_queue,
+                                  const std::string &image_id,
+                                  cls::rbd::MirrorImage *mirror_image,
+                                  PromotionState *promotion_state,
+                                  std::string* primary_mirror_uuid,
+                                  Context *on_finish)
+  : m_io_ctx(io_ctx), m_op_work_queue(op_work_queue), m_image_id(image_id),
+    m_mirror_image(mirror_image), m_promotion_state(promotion_state),
+    m_primary_mirror_uuid(primary_mirror_uuid), m_on_finish(on_finish),
+    m_cct(reinterpret_cast<CephContext *>(io_ctx.cct())) {
+}
+
+template <typename I>
+GetInfoRequest<I>::GetInfoRequest(I &image_ctx,
+                                  cls::rbd::MirrorImage *mirror_image,
+                                  PromotionState *promotion_state,
+                                  std::string* primary_mirror_uuid,
+                                  Context *on_finish)
+  : m_image_ctx(&image_ctx), m_io_ctx(image_ctx.md_ctx),
+    m_op_work_queue(image_ctx.op_work_queue), m_image_id(image_ctx.id),
+    m_mirror_image(mirror_image), m_promotion_state(promotion_state),
+    m_primary_mirror_uuid(primary_mirror_uuid), m_on_finish(on_finish),
+    m_cct(image_ctx.cct) {
+}
+
+template <typename I>
+void GetInfoRequest<I>::send() {
+  get_mirror_image();
+}
+
+template <typename I>
+void GetInfoRequest<I>::get_mirror_image() {
+  ldout(m_cct, 20) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_get_start(&op, m_image_id);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    GetInfoRequest<I>, &GetInfoRequest<I>::handle_get_mirror_image>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void GetInfoRequest<I>::handle_get_mirror_image(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  m_mirror_image->state = cls::rbd::MIRROR_IMAGE_STATE_DISABLED;
+  *m_promotion_state = PROMOTION_STATE_NON_PRIMARY;
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_image_get_finish(&iter, m_mirror_image);
+  }
+
+  if (r == -ENOENT) {
+    ldout(m_cct, 20) << "mirroring is disabled" << dendl;
+    finish(r);
+    return;
+  } else if (r < 0) {
+    lderr(m_cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_mirror_image->mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
+    get_journal_tag_owner();
+  } else if (m_mirror_image->mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    get_snapcontext();
+  } else {
+    ldout(m_cct, 20) << "unknown mirror image mode: " << m_mirror_image->mode
+                     << dendl;
+    finish(-EOPNOTSUPP);
+  }
+}
+
+template <typename I>
+void GetInfoRequest<I>::get_journal_tag_owner() {
+  ldout(m_cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    GetInfoRequest<I>, &GetInfoRequest<I>::handle_get_journal_tag_owner>(this);
+  Journal<I>::get_tag_owner(m_io_ctx, m_image_id, &m_mirror_uuid,
+                            m_op_work_queue, ctx);
+}
+
+template <typename I>
+void GetInfoRequest<I>::handle_get_journal_tag_owner(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to determine tag ownership: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_mirror_uuid == Journal<>::LOCAL_MIRROR_UUID) {
+    *m_promotion_state = PROMOTION_STATE_PRIMARY;
+    *m_primary_mirror_uuid = "";
+  } else if (m_mirror_uuid == Journal<>::ORPHAN_MIRROR_UUID) {
+    *m_promotion_state = PROMOTION_STATE_ORPHAN;
+    *m_primary_mirror_uuid = "";
+  } else {
+    *m_primary_mirror_uuid = m_mirror_uuid;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void GetInfoRequest<I>::get_snapcontext() {
+  if (m_image_ctx != nullptr) {
+    {
+      std::shared_lock image_locker{m_image_ctx->image_lock};
+      calc_promotion_state(m_image_ctx->snap_info);
+    }
+    finish(0);
+    return;
+  }
+
+  ldout(m_cct, 20) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::get_snapcontext_start(&op);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    GetInfoRequest<I>, &GetInfoRequest<I>::handle_get_snapcontext>(this);
+  m_out_bl.clear();
+  int r = m_io_ctx.aio_operate(util::header_name(m_image_id), comp, &op,
+                               &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void GetInfoRequest<I>::handle_get_snapcontext(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  if (r >= 0) {
+    auto it = m_out_bl.cbegin();
+    r = cls_client::get_snapcontext_finish(&it, &m_snapc);
+  }
+
+  if (r == -ENOENT &&
+      m_mirror_image->state == cls::rbd::MIRROR_IMAGE_STATE_CREATING) {
+    // image doesn't exist but we have a mirror image record for it
+    ldout(m_cct, 10) << "image does not exist for mirror image id "
+                     << m_image_id << dendl;
+    *m_promotion_state = PROMOTION_STATE_UNKNOWN;
+    *m_primary_mirror_uuid = "";
+    finish(0);
+    return;
+  } else if (r < 0) {
+    lderr(m_cct) << "failed to get snapcontext: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  get_snapshots();
+}
+
+
+template <typename I>
+void GetInfoRequest<I>::get_snapshots() {
+  ldout(m_cct, 20) << dendl;
+
+  if (m_snapc.snaps.empty()) {
+    handle_get_snapshots(0);
+    return;
+  }
+
+  librados::ObjectReadOperation op;
+  for (auto snap_id : m_snapc.snaps) {
+    cls_client::snapshot_get_start(&op, snap_id);
+  }
+
+  librados::AioCompletion *comp = create_rados_callback<
+    GetInfoRequest<I>, &GetInfoRequest<I>::handle_get_snapshots>(this);
+  m_out_bl.clear();
+  int r = m_io_ctx.aio_operate(util::header_name(m_image_id), comp, &op,
+                               &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void GetInfoRequest<I>::handle_get_snapshots(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  std::map<librados::snap_t, SnapInfo> snap_info;
+
+  auto it = m_out_bl.cbegin();
+  for (auto snap_id : m_snapc.snaps) {
+    cls::rbd::SnapshotInfo snap;
+    if (r >= 0) {
+      r = cls_client::snapshot_get_finish(&it, &snap);
+    }
+    snap_info.emplace(
+      snap_id, SnapInfo(snap.name, snap.snapshot_namespace, 0, {}, 0, 0, {}));
+  }
+
+  if (r == -ENOENT) {
+    // restart
+    get_snapcontext();
+    return;
+  }
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to get snapshots: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  calc_promotion_state(snap_info);
+  finish(0);
+}
+
+template <typename I>
+void GetInfoRequest<I>::finish(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+template <typename I>
+void GetInfoRequest<I>::calc_promotion_state(
+    const std::map<librados::snap_t, SnapInfo> &snap_info) {
+  *m_promotion_state = PROMOTION_STATE_UNKNOWN;
+  *m_primary_mirror_uuid = "";
+
+  for (auto it = snap_info.rbegin(); it != snap_info.rend(); it++) {
+    auto mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+      &it->second.snap_namespace);
+
+    if (mirror_ns != nullptr) {
+      switch (mirror_ns->state) {
+      case cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY:
+        *m_promotion_state = PROMOTION_STATE_PRIMARY;
+        break;
+      case cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY:
+        *m_promotion_state = PROMOTION_STATE_NON_PRIMARY;
+        *m_primary_mirror_uuid = mirror_ns->primary_mirror_uuid;
+        break;
+      case cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED:
+      case cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED:
+        *m_promotion_state = PROMOTION_STATE_ORPHAN;
+        break;
+      }
+      break;
+    }
+  }
+
+  ldout(m_cct, 10) << "promotion_state=" << *m_promotion_state << ", "
+                   << "primary_mirror_uuid=" << *m_primary_mirror_uuid << dendl;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::GetInfoRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/GetInfoRequest.h b/src/librbd/mirror/GetInfoRequest.h
new file mode 100644
index 000000000..dcc6da7da
--- /dev/null
+++ b/src/librbd/mirror/GetInfoRequest.h
@@ -0,0 +1,123 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_GET_INFO_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_GET_INFO_REQUEST_H
+
+#include "common/snap_types.h"
+#include "include/buffer.h"
+#include "include/common_fwd.h"
+#include "include/rados/librados.hpp"
+#include "librbd/Types.h"
+#include "librbd/mirror/Types.h"
+#include <string>
+
+struct Context;
+
+namespace cls { namespace rbd { struct MirrorImage; } }
+
+namespace librbd {
+
+struct ImageCtx;
+namespace asio { struct ContextWQ; }
+
+namespace mirror {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class GetInfoRequest {
+public:
+  static GetInfoRequest *create(librados::IoCtx &io_ctx,
+                                asio::ContextWQ *op_work_queue,
+                                const std::string &image_id,
+                                cls::rbd::MirrorImage *mirror_image,
+                                PromotionState *promotion_state,
+                                std::string* primary_mirror_uuid,
+                                Context *on_finish) {
+    return new GetInfoRequest(io_ctx, op_work_queue, image_id, mirror_image,
+                              promotion_state, primary_mirror_uuid, on_finish);
+  }
+  static GetInfoRequest *create(ImageCtxT &image_ctx,
+                                cls::rbd::MirrorImage *mirror_image,
+                                PromotionState *promotion_state,
+                                std::string* primary_mirror_uuid,
+                                Context *on_finish) {
+    return new GetInfoRequest(image_ctx, mirror_image, promotion_state,
+                              primary_mirror_uuid, on_finish);
+  }
+
+  GetInfoRequest(librados::IoCtx& io_ctx, asio::ContextWQ *op_work_queue,
+                 const std::string &image_id,
+                 cls::rbd::MirrorImage *mirror_image,
+                 PromotionState *promotion_state,
+                 std::string* primary_mirror_uuid, Context *on_finish);
+  GetInfoRequest(ImageCtxT &image_ctx, cls::rbd::MirrorImage *mirror_image,
+                 PromotionState *promotion_state,
+                 std::string* primary_mirror_uuid, Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   *                  <start>
+   *                     |
+   *                     v
+   *              GET_MIRROR_IMAGE
+   *                     |
+   *  (journal /--------/ \--------\ (snapshot
+   *   mode)   |                   |  mode)
+   *           v                   v
+   *  GET_JOURNAL_TAG_OWNER    GET_SNAPCONTEXT (skip if
+   *           |                   |            cached)
+   *           |                   v
+   *           |               GET_SNAPSHOTS (skip if
+   *           |                   |          cached)
+   *           \--------\ /--------/
+   *                     |
+   *                     v
+   *                  <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx = nullptr;
+  librados::IoCtx &m_io_ctx;
+  asio::ContextWQ *m_op_work_queue;
+  std::string m_image_id;
+  cls::rbd::MirrorImage *m_mirror_image;
+  PromotionState *m_promotion_state;
+  std::string* m_primary_mirror_uuid;
+  Context *m_on_finish;
+
+  CephContext *m_cct;
+
+  bufferlist m_out_bl;
+  std::string m_mirror_uuid;
+  ::SnapContext m_snapc;
+
+  void get_mirror_image();
+  void handle_get_mirror_image(int r);
+
+  void get_journal_tag_owner();
+  void handle_get_journal_tag_owner(int r);
+
+  void get_snapcontext();
+  void handle_get_snapcontext(int r);
+
+  void get_snapshots();
+  void handle_get_snapshots(int r);
+
+  void finish(int r);
+
+  void calc_promotion_state(
+    const std::map<librados::snap_t, SnapInfo> &snap_info);
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::GetInfoRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_GET_INFO_REQUEST_H
+
diff --git a/src/librbd/mirror/GetStatusRequest.cc b/src/librbd/mirror/GetStatusRequest.cc
new file mode 100644
index 000000000..40d4a664b
--- /dev/null
+++ b/src/librbd/mirror/GetStatusRequest.cc
@@ -0,0 +1,116 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/GetStatusRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/GetInfoRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::GetStatusRequest: " << this \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void GetStatusRequest<I>::send() {
+  *m_mirror_image_status = cls::rbd::MirrorImageStatus(
+    {{cls::rbd::MirrorImageSiteStatus::LOCAL_MIRROR_UUID,
+      cls::rbd::MIRROR_IMAGE_STATUS_STATE_UNKNOWN, "status not found"}});
+
+  get_info();
+}
+
+template <typename I>
+void GetStatusRequest<I>::get_info() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    GetStatusRequest<I>, &GetStatusRequest<I>::handle_get_info>(this);
+  auto req = GetInfoRequest<I>::create(m_image_ctx, m_mirror_image,
+                                       m_promotion_state,
+                                       &m_primary_mirror_uuid, ctx);
+  req->send();
+}
+
+template <typename I>
+void GetStatusRequest<I>::handle_get_info(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    if (r != -ENOENT) {
+      lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+                 << dendl;
+    }
+    finish(r);
+    return;
+  } else if (m_mirror_image->state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+    finish(0);
+    return;
+  }
+
+  get_status();
+}
+
+template <typename I>
+void GetStatusRequest<I>::get_status() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_status_get_start(
+    &op, m_mirror_image->global_image_id);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    GetStatusRequest<I>, &GetStatusRequest<I>::handle_get_status>(this);
+  int r = m_image_ctx.md_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void GetStatusRequest<I>::handle_get_status(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_image_status_get_finish(&iter,
+                                                   m_mirror_image_status);
+  }
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(cct) << "failed to retrieve mirror image status: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void GetStatusRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::GetStatusRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/GetStatusRequest.h b/src/librbd/mirror/GetStatusRequest.h
new file mode 100644
index 000000000..581a0d667
--- /dev/null
+++ b/src/librbd/mirror/GetStatusRequest.h
@@ -0,0 +1,86 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_GET_STATUS_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_GET_STATUS_REQUEST_H
+
+#include "include/buffer.h"
+#include "librbd/mirror/Types.h"
+#include <string>
+
+struct Context;
+namespace cls { namespace rbd { struct MirrorImage; } }
+namespace cls { namespace rbd { struct MirrorImageStatus; } }
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class GetStatusRequest {
+public:
+  static GetStatusRequest *create(ImageCtxT &image_ctx,
+                                  cls::rbd::MirrorImageStatus *status,
+                                  cls::rbd::MirrorImage *mirror_image,
+                                  PromotionState *promotion_state,
+                                  Context *on_finish) {
+    return new GetStatusRequest(image_ctx, status, mirror_image,
+                                promotion_state, on_finish);
+  }
+
+  GetStatusRequest(ImageCtxT &image_ctx, cls::rbd::MirrorImageStatus *status,
+                   cls::rbd::MirrorImage *mirror_image,
+                   PromotionState *promotion_state, Context *on_finish)
+    : m_image_ctx(image_ctx), m_mirror_image_status(status),
+      m_mirror_image(mirror_image), m_promotion_state(promotion_state),
+      m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_INFO
+   *    |
+   *    v
+   * GET_STATUS
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT &m_image_ctx;
+  cls::rbd::MirrorImageStatus *m_mirror_image_status;
+  cls::rbd::MirrorImage *m_mirror_image;
+  PromotionState *m_promotion_state;
+  Context *m_on_finish;
+
+  bufferlist m_out_bl;
+  std::string m_primary_mirror_uuid;
+
+  void get_info();
+  void handle_get_info(int r);
+
+  void get_status();
+  void handle_get_status(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::GetStatusRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_GET_STATUS_REQUEST_H
+
diff --git a/src/librbd/mirror/GetUuidRequest.cc b/src/librbd/mirror/GetUuidRequest.cc
new file mode 100644
index 000000000..f8209f905
--- /dev/null
+++ b/src/librbd/mirror/GetUuidRequest.cc
@@ -0,0 +1,86 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/GetUuidRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::GetUuidRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using librbd::util::create_rados_callback;
+
+template <typename I>
+GetUuidRequest<I>::GetUuidRequest(
+    librados::IoCtx& io_ctx, std::string* mirror_uuid, Context* on_finish)
+  : m_mirror_uuid(mirror_uuid), m_on_finish(on_finish),
+    m_cct(reinterpret_cast<CephContext*>(io_ctx.cct())) {
+  m_io_ctx.dup(io_ctx);
+  m_io_ctx.set_namespace("");
+}
+
+template <typename I>
+void GetUuidRequest<I>::send() {
+  get_mirror_uuid();
+}
+
+template <typename I>
+void GetUuidRequest<I>::get_mirror_uuid() {
+  ldout(m_cct, 20) << dendl;
+
+  librados::ObjectReadOperation op;
+  librbd::cls_client::mirror_uuid_get_start(&op);
+
+  auto aio_comp = create_rados_callback<
+    GetUuidRequest<I>, &GetUuidRequest<I>::handle_get_mirror_uuid>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void GetUuidRequest<I>::handle_get_mirror_uuid(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  if (r >= 0) {
+    auto it = m_out_bl.cbegin();
+    r = librbd::cls_client::mirror_uuid_get_finish(&it, m_mirror_uuid);
+    if (r >= 0 && m_mirror_uuid->empty()) {
+      r = -ENOENT;
+    }
+  }
+
+  if (r < 0) {
+    if (r == -ENOENT) {
+      ldout(m_cct, 5) << "mirror uuid missing" << dendl;
+    } else {
+      lderr(m_cct) << "failed to retrieve mirror uuid: " << cpp_strerror(r)
+                   << dendl;
+    }
+    *m_mirror_uuid = "";
+  }
+
+  finish(r);
+}
+
+template <typename I>
+void GetUuidRequest<I>::finish(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::GetUuidRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/GetUuidRequest.h b/src/librbd/mirror/GetUuidRequest.h
new file mode 100644
index 000000000..73cc2d5b2
--- /dev/null
+++ b/src/librbd/mirror/GetUuidRequest.h
@@ -0,0 +1,69 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_GET_UUID_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_GET_UUID_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/rados/librados.hpp"
+#include "cls/rbd/cls_rbd_types.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class GetUuidRequest {
+public:
+  static GetUuidRequest *create(librados::IoCtx& io_ctx,
+                                std::string* mirror_uuid, Context* on_finish) {
+    return new GetUuidRequest(io_ctx, mirror_uuid, on_finish);
+  }
+
+  GetUuidRequest(librados::IoCtx& io_ctx, std::string* mirror_uuid,
+                 Context* on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_MIRROR_UUID
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  librados::IoCtx m_io_ctx;
+  std::string* m_mirror_uuid;
+  Context* m_on_finish;
+
+  CephContext* m_cct;
+
+  bufferlist m_out_bl;
+
+  void get_mirror_uuid();
+  void handle_get_mirror_uuid(int r);
+
+  void finish(int r);
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::GetUuidRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_GET_UUID_REQUEST_H
diff --git a/src/librbd/mirror/ImageRemoveRequest.cc b/src/librbd/mirror/ImageRemoveRequest.cc
new file mode 100644
index 000000000..1aa265dae
--- /dev/null
+++ b/src/librbd/mirror/ImageRemoveRequest.cc
@@ -0,0 +1,98 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/ImageRemoveRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/MirroringWatcher.h"
+#include "librbd/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::ImageRemoveRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using util::create_rados_callback;
+
+template <typename I>
+ImageRemoveRequest<I>::ImageRemoveRequest(
+    librados::IoCtx& io_ctx, const std::string& global_image_id,
+    const std::string& image_id, Context* on_finish)
+  : m_io_ctx(io_ctx), m_global_image_id(global_image_id), m_image_id(image_id),
+    m_on_finish(on_finish), m_cct(static_cast<CephContext*>(m_io_ctx.cct())) {
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::send() {
+  remove_mirror_image();
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::remove_mirror_image() {
+  ldout(m_cct, 10) << dendl;
+
+  librados::ObjectWriteOperation op;
+  cls_client::mirror_image_remove(&op, m_image_id);
+
+  auto comp = create_rados_callback<
+    ImageRemoveRequest<I>,
+    &ImageRemoveRequest<I>::handle_remove_mirror_image>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::handle_remove_mirror_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(m_cct) << "failed to remove mirroring image: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  notify_mirroring_watcher();
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::notify_mirroring_watcher() {
+  ldout(m_cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    ImageRemoveRequest<I>,
+    &ImageRemoveRequest<I>::handle_notify_mirroring_watcher>(this);
+  MirroringWatcher<I>::notify_image_updated(
+    m_io_ctx, cls::rbd::MIRROR_IMAGE_STATE_DISABLED,
+    m_image_id, m_global_image_id, ctx);
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::handle_notify_mirroring_watcher(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to notify mirror image update: " << cpp_strerror(r)
+                 << dendl;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void ImageRemoveRequest<I>::finish(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::ImageRemoveRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/ImageRemoveRequest.h b/src/librbd/mirror/ImageRemoveRequest.h
new file mode 100644
index 000000000..c04f9fadc
--- /dev/null
+++ b/src/librbd/mirror/ImageRemoveRequest.h
@@ -0,0 +1,77 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_IMAGE_REMOVE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_IMAGE_REMOVE_REQUEST_H
+
+#include "include/rados/librados.hpp"
+#include "common/ceph_mutex.h"
+#include "cls/rbd/cls_rbd_types.h"
+
+#include <string>
+
+class Context;
+
+namespace librbd {
+
+class ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = ImageCtx>
+class ImageRemoveRequest {
+public:
+  static ImageRemoveRequest *create(librados::IoCtx& io_ctx,
+                                    const std::string& global_image_id,
+                                    const std::string& image_id,
+                                    Context* on_finish) {
+    return new ImageRemoveRequest(io_ctx, global_image_id, image_id, on_finish);
+  }
+
+  ImageRemoveRequest(librados::IoCtx& io_ctx,
+                     const std::string& global_image_id,
+                     const std::string& image_id,
+                     Context* on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   * REMOVE_MIRROR_IMAGE
+   *    |
+   *    v
+   * NOTIFY_MIRRORING_WATCHER
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  librados::IoCtx& m_io_ctx;
+  std::string m_global_image_id;
+  std::string m_image_id;
+  Context* m_on_finish;
+
+  CephContext* m_cct;
+
+  void remove_mirror_image();
+  void handle_remove_mirror_image(int r);
+
+  void notify_mirroring_watcher();
+  void handle_notify_mirroring_watcher(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::ImageRemoveRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_IMAGE_REMOVE_REQUEST_H
diff --git a/src/librbd/mirror/ImageStateUpdateRequest.cc b/src/librbd/mirror/ImageStateUpdateRequest.cc
new file mode 100644
index 000000000..98e987190
--- /dev/null
+++ b/src/librbd/mirror/ImageStateUpdateRequest.cc
@@ -0,0 +1,151 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/ImageStateUpdateRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/MirroringWatcher.h"
+#include "librbd/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::ImageStateUpdateRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using util::create_rados_callback;
+
+template <typename I>
+ImageStateUpdateRequest<I>::ImageStateUpdateRequest(
+    librados::IoCtx& io_ctx,
+    const std::string& image_id,
+    cls::rbd::MirrorImageState mirror_image_state,
+    const cls::rbd::MirrorImage& mirror_image,
+    Context* on_finish)
+  : m_io_ctx(io_ctx), m_image_id(image_id),
+    m_mirror_image_state(mirror_image_state), m_mirror_image(mirror_image),
+    m_on_finish(on_finish), m_cct(static_cast<CephContext*>(m_io_ctx.cct())) {
+  ceph_assert(m_mirror_image_state != cls::rbd::MIRROR_IMAGE_STATE_DISABLED);
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::send() {
+  get_mirror_image();
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::get_mirror_image() {
+  if (!m_mirror_image.global_image_id.empty()) {
+    set_mirror_image();
+    return;
+  }
+
+  ldout(m_cct, 10) << dendl;
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_get_start(&op, m_image_id);
+
+  auto comp = create_rados_callback<
+    ImageStateUpdateRequest<I>,
+    &ImageStateUpdateRequest<I>::handle_get_mirror_image>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::handle_get_mirror_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_image_get_finish(&iter, &m_mirror_image);
+  }
+
+  if (r == -ENOENT) {
+    ldout(m_cct, 20) << "mirroring is disabled" << dendl;
+    finish(0);
+    return;
+  } else if (r < 0) {
+    lderr(m_cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  set_mirror_image();
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::set_mirror_image() {
+  if (m_mirror_image.state == m_mirror_image_state) {
+    finish(0);
+    return;
+  }
+
+  ldout(m_cct, 10) << dendl;
+  m_mirror_image.state = m_mirror_image_state;
+
+  librados::ObjectWriteOperation op;
+  cls_client::mirror_image_set(&op, m_image_id, m_mirror_image);
+
+  auto comp = create_rados_callback<
+    ImageStateUpdateRequest<I>,
+    &ImageStateUpdateRequest<I>::handle_set_mirror_image>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::handle_set_mirror_image(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to disable mirroring image: " << cpp_strerror(r)
+                 << dendl;
+    finish(r);
+    return;
+  }
+
+  notify_mirroring_watcher();
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::notify_mirroring_watcher() {
+  ldout(m_cct, 10) << dendl;
+
+  auto ctx = util::create_context_callback<
+    ImageStateUpdateRequest<I>,
+    &ImageStateUpdateRequest<I>::handle_notify_mirroring_watcher>(this);
+  MirroringWatcher<I>::notify_image_updated(
+    m_io_ctx, m_mirror_image_state, m_image_id, m_mirror_image.global_image_id,
+    ctx);
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::handle_notify_mirroring_watcher(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_cct) << "failed to notify mirror image update: " << cpp_strerror(r)
+                 << dendl;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void ImageStateUpdateRequest<I>::finish(int r) {
+  ldout(m_cct, 10) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::ImageStateUpdateRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/ImageStateUpdateRequest.h b/src/librbd/mirror/ImageStateUpdateRequest.h
new file mode 100644
index 000000000..9e0affe6a
--- /dev/null
+++ b/src/librbd/mirror/ImageStateUpdateRequest.h
@@ -0,0 +1,92 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_IMAGE_STATE_UPDATE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_IMAGE_STATE_UPDATE_REQUEST_H
+
+#include "include/rados/librados.hpp"
+#include "common/ceph_mutex.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/mirror/Types.h"
+
+#include <string>
+
+class Context;
+
+namespace librbd {
+
+class ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = ImageCtx>
+class ImageStateUpdateRequest {
+public:
+  static ImageStateUpdateRequest *create(
+      librados::IoCtx& io_ctx,
+      const std::string& image_id,
+      cls::rbd::MirrorImageState mirror_image_state,
+      const cls::rbd::MirrorImage& mirror_image,
+      Context* on_finish) {
+    return new ImageStateUpdateRequest(
+      io_ctx, image_id, mirror_image_state, mirror_image, on_finish);
+  }
+
+  ImageStateUpdateRequest(
+      librados::IoCtx& io_ctx,
+      const std::string& image_id,
+      cls::rbd::MirrorImageState mirror_image_state,
+      const cls::rbd::MirrorImage& mirror_image,
+      Context* on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v (skip if provided)
+   * GET_MIRROR_IMAGE
+   *    |
+   *    v
+   * SET_MIRROR_IMAGE
+   *    |
+   *    v
+   * NOTIFY_MIRRORING_WATCHER
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  librados::IoCtx& m_io_ctx;
+  std::string m_image_id;
+  cls::rbd::MirrorImageState m_mirror_image_state;
+  cls::rbd::MirrorImage m_mirror_image;
+  Context* m_on_finish;
+
+  CephContext* m_cct;
+  bufferlist m_out_bl;
+
+  void get_mirror_image();
+  void handle_get_mirror_image(int r);
+
+  void set_mirror_image();
+  void handle_set_mirror_image(int r);
+
+  void notify_mirroring_watcher();
+  void handle_notify_mirroring_watcher(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::ImageStateUpdateRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_IMAGE_STATE_UPDATE_REQUEST_H
diff --git a/src/librbd/mirror/PromoteRequest.cc b/src/librbd/mirror/PromoteRequest.cc
new file mode 100644
index 000000000..b119e4edc
--- /dev/null
+++ b/src/librbd/mirror/PromoteRequest.cc
@@ -0,0 +1,115 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/PromoteRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/GetInfoRequest.h"
+#include "librbd/mirror/snapshot/PromoteRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::PromoteRequest: " << this \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+
+using librbd::util::create_context_callback;
+
+template <typename I>
+void PromoteRequest<I>::send() {
+  get_info();
+}
+
+template <typename I>
+void PromoteRequest<I>::get_info() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>, &PromoteRequest<I>::handle_get_info>(this);
+  auto req = GetInfoRequest<I>::create(m_image_ctx, &m_mirror_image,
+                                       &m_promotion_state,
+                                       &m_primary_mirror_uuid, ctx);
+  req->send();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_get_info(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  } else if (m_mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+    lderr(cct) << "mirroring is not currently enabled" << dendl;
+    finish(-EINVAL);
+    return;
+  } else if (m_promotion_state == PROMOTION_STATE_PRIMARY) {
+    lderr(cct) << "image is already primary" << dendl;
+    finish(-EINVAL);
+    return;
+  } else if (m_promotion_state == PROMOTION_STATE_NON_PRIMARY && !m_force) {
+    lderr(cct) << "image is primary within a remote cluster or demotion is not propagated yet"
+               << dendl;
+    finish(-EBUSY);
+    return;
+  }
+
+  promote();
+}
+
+template <typename I>
+void PromoteRequest<I>::promote() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>, &PromoteRequest<I>::handle_promote>(this);
+  if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
+    Journal<I>::promote(&m_image_ctx, ctx);
+  } else if (m_mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    auto req = mirror::snapshot::PromoteRequest<I>::create(
+      &m_image_ctx, m_mirror_image.global_image_id, ctx);
+    req->send();
+  } else {
+    lderr(cct) << "unknown image mirror mode: " << m_mirror_image.mode << dendl;
+    finish(-EOPNOTSUPP);
+  }
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_promote(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to promote image: " << cpp_strerror(r)
+               << dendl;
+  }
+
+  finish(r);
+}
+
+template <typename I>
+void PromoteRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::PromoteRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/PromoteRequest.h b/src/librbd/mirror/PromoteRequest.h
new file mode 100644
index 000000000..c54f3bb76
--- /dev/null
+++ b/src/librbd/mirror/PromoteRequest.h
@@ -0,0 +1,76 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_PROMOTE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_PROMOTE_REQUEST_H
+
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/mirror/Types.h"
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class PromoteRequest {
+public:
+  static PromoteRequest *create(ImageCtxT &image_ctx, bool force,
+                                Context *on_finish) {
+    return new PromoteRequest(image_ctx, force, on_finish);
+  }
+
+  PromoteRequest(ImageCtxT &image_ctx, bool force, Context *on_finish)
+    : m_image_ctx(image_ctx), m_force(force), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_INFO
+   *    |
+   *    v
+   * GET_TAG_OWNER
+   *    |
+   *    v
+   * PROMOTE
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT &m_image_ctx;
+  bool m_force;
+  Context *m_on_finish;
+
+  cls::rbd::MirrorImage m_mirror_image;
+  PromotionState m_promotion_state = PROMOTION_STATE_PRIMARY;
+  std::string m_primary_mirror_uuid;
+
+  void get_info();
+  void handle_get_info(int r);
+
+  void promote();
+  void handle_promote(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::PromoteRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_PROMOTE_REQUEST_H
diff --git a/src/librbd/mirror/Types.h b/src/librbd/mirror/Types.h
new file mode 100644
index 000000000..2388b74ef
--- /dev/null
+++ b/src/librbd/mirror/Types.h
@@ -0,0 +1,21 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_TYPES_H
+#define CEPH_LIBRBD_MIRROR_TYPES_H
+
+namespace librbd {
+namespace mirror {
+
+enum PromotionState {
+  PROMOTION_STATE_UNKNOWN,
+  PROMOTION_STATE_PRIMARY,
+  PROMOTION_STATE_NON_PRIMARY,
+  PROMOTION_STATE_ORPHAN
+};
+
+} // namespace mirror
+} // namespace librbd
+
+#endif // CEPH_LIBRBD_MIRROR_TYPES_H
+
diff --git a/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc
new file mode 100644
index 000000000..eed0aa506
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc
@@ -0,0 +1,273 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/CreateNonPrimaryRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/Utils.h"
+#include "librbd/mirror/snapshot/WriteImageStateRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::CreateNonPrimaryRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+CreateNonPrimaryRequest<I>::CreateNonPrimaryRequest(
+    I* image_ctx, bool demoted, const std::string &primary_mirror_uuid,
+    uint64_t primary_snap_id, const SnapSeqs& snap_seqs,
+    const ImageState &image_state, uint64_t *snap_id, Context *on_finish)
+  : m_image_ctx(image_ctx), m_demoted(demoted),
+    m_primary_mirror_uuid(primary_mirror_uuid),
+    m_primary_snap_id(primary_snap_id), m_snap_seqs(snap_seqs),
+    m_image_state(image_state), m_snap_id(snap_id), m_on_finish(on_finish) {
+  m_default_ns_ctx.dup(m_image_ctx->md_ctx);
+  m_default_ns_ctx.set_namespace("");
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::send() {
+  refresh_image();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::refresh_image() {
+  if (!m_image_ctx->state->is_refresh_required()) {
+    get_mirror_image();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_refresh_image>(this);
+  m_image_ctx->state->refresh(ctx);
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_refresh_image(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to refresh image: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  get_mirror_image();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::get_mirror_image() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_get_start(&op, m_image_ctx->id);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_get_mirror_image>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_get_mirror_image(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  cls::rbd::MirrorImage mirror_image;
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_image_get_finish(&iter, &mirror_image);
+  }
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (mirror_image.mode != cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
+    lderr(cct) << "snapshot based mirroring is not enabled" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  if (!is_orphan() && !util::can_create_non_primary_snapshot(m_image_ctx)) {
+    finish(-EINVAL);
+    return;
+  }
+
+  uuid_d uuid_gen;
+  uuid_gen.generate_random();
+  m_snap_name = ".mirror.non_primary." + mirror_image.global_image_id + "." +
+    uuid_gen.to_string();
+
+  get_mirror_peers();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::get_mirror_peers() {
+  if (!m_demoted) {
+    create_snapshot();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_peer_list_start(&op);
+
+  auto aio_comp = create_rados_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_get_mirror_peers>(this);
+  m_out_bl.clear();
+  int r = m_default_ns_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_get_mirror_peers(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  std::vector<cls::rbd::MirrorPeer> peers;
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_peer_list_finish(&iter, &peers);
+  }
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve mirror peers: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  for (auto &peer : peers) {
+    if (peer.mirror_peer_direction == cls::rbd::MIRROR_PEER_DIRECTION_RX) {
+      continue;
+    }
+    m_mirror_peer_uuids.insert(peer.uuid);
+  }
+
+  create_snapshot();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::create_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+
+  cls::rbd::MirrorSnapshotNamespace ns{
+    (m_demoted ? cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED :
+                 cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY), {},
+    m_primary_mirror_uuid, m_primary_snap_id};
+  if (m_demoted) {
+    ns.mirror_peer_uuids = m_mirror_peer_uuids;
+  }
+  ns.snap_seqs = m_snap_seqs;
+  ns.complete = is_orphan();
+  ldout(cct, 15) << "ns=" << ns << dendl;
+
+  auto ctx = create_context_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_create_snapshot>(this);
+  m_image_ctx->operations->snap_create(ns, m_snap_name, 0, m_prog_ctx, ctx);
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_create_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create mirror snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  write_image_state();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::write_image_state() {
+  uint64_t snap_id;
+  {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+    snap_id = m_image_ctx->get_snap_id(
+      cls::rbd::MirrorSnapshotNamespace{}, m_snap_name);
+  }
+
+  if (m_snap_id != nullptr) {
+    *m_snap_id = snap_id;
+  }
+
+  if (is_orphan()) {
+    finish(0);
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_write_image_state>(this);
+
+  auto req = WriteImageStateRequest<I>::create(m_image_ctx, snap_id,
+                                               m_image_state, ctx);
+  req->send();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_write_image_state(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to write image state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::CreateNonPrimaryRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.h b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.h
new file mode 100644
index 000000000..36f155413
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.h
@@ -0,0 +1,123 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_NON_PRIMARY_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_NON_PRIMARY_REQUEST_H
+
+#include "include/buffer.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/Types.h"
+#include "librbd/internal.h"
+#include "librbd/mirror/snapshot/Types.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class CreateNonPrimaryRequest {
+public:
+  static CreateNonPrimaryRequest *create(ImageCtxT *image_ctx,
+                                         bool demoted,
+                                         const std::string &primary_mirror_uuid,
+                                         uint64_t primary_snap_id,
+                                         const SnapSeqs& snap_seqs,
+                                         const ImageState &image_state,
+                                         uint64_t *snap_id,
+                                         Context *on_finish) {
+    return new CreateNonPrimaryRequest(image_ctx, demoted, primary_mirror_uuid,
+                                       primary_snap_id, snap_seqs, image_state,
+                                       snap_id, on_finish);
+  }
+
+  CreateNonPrimaryRequest(ImageCtxT *image_ctx,
+                          bool demoted,
+                          const std::string &primary_mirror_uuid,
+                          uint64_t primary_snap_id,
+                          const SnapSeqs& snap_seqs,
+                          const ImageState &image_state, uint64_t *snap_id,
+                          Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * REFRESH_IMAGE
+   *    |
+   *    v
+   * GET_MIRROR_IMAGE
+   *    |
+   *    v (skip if not needed)
+   * GET_MIRROR_PEERS
+   *    |
+   *    v
+   * CREATE_SNAPSHOT
+   *    |
+   *    v
+   * WRITE_IMAGE_STATE
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  bool m_demoted;
+  std::string m_primary_mirror_uuid;
+  uint64_t m_primary_snap_id;
+  SnapSeqs m_snap_seqs;
+  ImageState m_image_state;
+  uint64_t *m_snap_id;
+  Context *m_on_finish;
+
+  librados::IoCtx m_default_ns_ctx;
+  std::set<std::string> m_mirror_peer_uuids;
+
+  std::string m_snap_name;
+
+  bufferlist m_out_bl;
+  NoOpProgressContext m_prog_ctx;
+
+  bool is_orphan() const {
+    return m_primary_mirror_uuid.empty();
+  }
+
+  void refresh_image();
+  void handle_refresh_image(int r);
+
+  void get_mirror_image();
+  void handle_get_mirror_image(int r);
+
+  void get_mirror_peers();
+  void handle_get_mirror_peers(int r);
+
+  void create_snapshot();
+  void handle_create_snapshot(int r);
+
+  void write_image_state();
+  void handle_write_image_state(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::CreateNonPrimaryRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_NON_PRIMARY_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/CreatePrimaryRequest.cc b/src/librbd/mirror/snapshot/CreatePrimaryRequest.cc
new file mode 100644
index 000000000..54da9ad61
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreatePrimaryRequest.cc
@@ -0,0 +1,277 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/UnlinkPeerRequest.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::CreatePrimaryRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+CreatePrimaryRequest<I>::CreatePrimaryRequest(
+    I *image_ctx, const std::string& global_image_id,
+    uint64_t clean_since_snap_id, uint64_t snap_create_flags, uint32_t flags,
+    uint64_t *snap_id, Context *on_finish)
+  : m_image_ctx(image_ctx), m_global_image_id(global_image_id),
+    m_clean_since_snap_id(clean_since_snap_id),
+    m_snap_create_flags(snap_create_flags), m_flags(flags), m_snap_id(snap_id),
+    m_on_finish(on_finish) {
+  m_default_ns_ctx.dup(m_image_ctx->md_ctx);
+  m_default_ns_ctx.set_namespace("");
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::send() {
+  if (!util::can_create_primary_snapshot(
+        m_image_ctx,
+        ((m_flags & CREATE_PRIMARY_FLAG_DEMOTED) != 0),
+        ((m_flags & CREATE_PRIMARY_FLAG_FORCE) != 0), nullptr, nullptr)) {
+    finish(-EINVAL);
+    return;
+  }
+
+  uuid_d uuid_gen;
+  uuid_gen.generate_random();
+  m_snap_name = ".mirror.primary." + m_global_image_id + "." +
+    uuid_gen.to_string();
+
+  get_mirror_peers();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::get_mirror_peers() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_peer_list_start(&op);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<I>::handle_get_mirror_peers>(this);
+  m_out_bl.clear();
+  int r = m_default_ns_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_out_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::handle_get_mirror_peers(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  std::vector<cls::rbd::MirrorPeer> peers;
+  if (r == 0) {
+    auto iter = m_out_bl.cbegin();
+    r = cls_client::mirror_peer_list_finish(&iter, &peers);
+  }
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve mirror peers: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  for (auto &peer : peers) {
+    if (peer.mirror_peer_direction == cls::rbd::MIRROR_PEER_DIRECTION_RX) {
+      continue;
+    }
+    m_mirror_peer_uuids.insert(peer.uuid);
+  }
+
+  if (m_mirror_peer_uuids.empty() &&
+      ((m_flags & CREATE_PRIMARY_FLAG_IGNORE_EMPTY_PEERS) == 0)) {
+    lderr(cct) << "no mirror tx peers configured for the pool" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  create_snapshot();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::create_snapshot() {
+  cls::rbd::MirrorSnapshotNamespace ns{
+    ((m_flags & CREATE_PRIMARY_FLAG_DEMOTED) != 0 ?
+      cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED :
+      cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY),
+    m_mirror_peer_uuids, "", m_clean_since_snap_id};
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "name=" << m_snap_name << ", "
+                 << "ns=" << ns << dendl;
+  auto ctx = create_context_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<I>::handle_create_snapshot>(this);
+  m_image_ctx->operations->snap_create(ns, m_snap_name, m_snap_create_flags,
+                                       m_prog_ctx, ctx);
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::handle_create_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create mirror snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  refresh_image();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::refresh_image() {
+  // if snapshot created via remote RPC, refresh is required to retrieve
+  // the snapshot id
+  if (m_snap_id == nullptr) {
+    unlink_peer();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<I>::handle_refresh_image>(this);
+  m_image_ctx->state->refresh(ctx);
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::handle_refresh_image(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to refresh image: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+    *m_snap_id = m_image_ctx->get_snap_id(
+      cls::rbd::MirrorSnapshotNamespace{}, m_snap_name);
+    ldout(cct, 15) << "snap_id=" << *m_snap_id << dendl;
+  }
+
+  unlink_peer();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::unlink_peer() {
+  uint64_t max_snapshots = m_image_ctx->config.template get_val<uint64_t>(
+    "rbd_mirroring_max_mirroring_snapshots");
+  ceph_assert(max_snapshots >= 3);
+
+  std::string peer_uuid;
+  uint64_t snap_id = CEPH_NOSNAP;
+
+  for (auto &peer : m_mirror_peer_uuids) {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+    size_t count = 0;
+    uint64_t unlink_snap_id = 0;
+    for (auto &snap_it : m_image_ctx->snap_info) {
+      auto info = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+        &snap_it.second.snap_namespace);
+      if (info == nullptr) {
+        continue;
+      }
+      if (info->state != cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY) {
+        // reset counters -- we count primary snapshots after the last promotion
+        count = 0;
+        unlink_snap_id = 0;
+        continue;
+      }
+      // call UnlinkPeerRequest only if the snapshot is linked with this peer
+      // or if it's not linked with any peer (happens if mirroring is enabled
+      // on a pool with no peers configured or if UnlinkPeerRequest gets
+      // interrupted)
+      if (info->mirror_peer_uuids.size() == 0) {
+        peer_uuid = peer;
+        snap_id = snap_it.first;
+        break;
+      }
+      if (info->mirror_peer_uuids.count(peer) == 0) {
+        continue;
+      }
+      count++;
+      if (count == max_snapshots) {
+        unlink_snap_id = snap_it.first;
+      }
+      if (count > max_snapshots) {
+        peer_uuid = peer;
+        snap_id = unlink_snap_id;
+        break;
+      }
+    }
+    if (snap_id != CEPH_NOSNAP) {
+      break;
+    }
+  }
+
+  if (snap_id == CEPH_NOSNAP) {
+    finish(0);
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "peer=" << peer_uuid << ", snap_id=" << snap_id << dendl;
+
+  auto ctx = create_context_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<I>::handle_unlink_peer>(this);
+  auto req = UnlinkPeerRequest<I>::create(m_image_ctx, snap_id, peer_uuid, ctx);
+  req->send();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::handle_unlink_peer(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to unlink peer: " << cpp_strerror(r) << dendl;
+    finish(0); // not fatal
+    return;
+  }
+
+  unlink_peer();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::CreatePrimaryRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/CreatePrimaryRequest.h b/src/librbd/mirror/snapshot/CreatePrimaryRequest.h
new file mode 100644
index 000000000..b8e84cf2b
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreatePrimaryRequest.h
@@ -0,0 +1,106 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_PRIMARY_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_PRIMARY_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/rados/librados.hpp"
+#include "cls/rbd/cls_rbd_types.h"
+#include "librbd/internal.h"
+#include "librbd/mirror/snapshot/Types.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class CreatePrimaryRequest {
+public:
+  static CreatePrimaryRequest *create(ImageCtxT *image_ctx,
+                                      const std::string& global_image_id,
+                                      uint64_t clean_since_snap_id,
+                                      uint64_t snap_create_flags,
+                                      uint32_t flags, uint64_t *snap_id,
+                                      Context *on_finish) {
+    return new CreatePrimaryRequest(image_ctx, global_image_id,
+                                    clean_since_snap_id, snap_create_flags, flags,
+                                    snap_id, on_finish);
+  }
+
+  CreatePrimaryRequest(ImageCtxT *image_ctx,
+                       const std::string& global_image_id,
+                       uint64_t clean_since_snap_id, uint64_t snap_create_flags,
+                       uint32_t flags, uint64_t *snap_id, Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_MIRROR_PEERS
+   *    |
+   *    v
+   * CREATE_SNAPSHOT
+   *    |
+   *    v
+   * REFRESH_IMAGE
+   *    |
+   *    v
+   * UNLINK_PEER (skip if not needed,
+   *    |         repeat if needed)
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  std::string m_global_image_id;
+  uint64_t m_clean_since_snap_id;
+  const uint64_t m_snap_create_flags;
+  const uint32_t m_flags;
+  uint64_t *m_snap_id;
+  Context *m_on_finish;
+
+  librados::IoCtx m_default_ns_ctx;
+  std::set<std::string> m_mirror_peer_uuids;
+  std::string m_snap_name;
+
+  bufferlist m_out_bl;
+  NoOpProgressContext m_prog_ctx;
+
+  void get_mirror_peers();
+  void handle_get_mirror_peers(int r);
+
+  void create_snapshot();
+  void handle_create_snapshot(int r);
+
+  void refresh_image();
+  void handle_refresh_image(int r);
+
+  void unlink_peer();
+  void handle_unlink_peer(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::CreatePrimaryRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_CREATE_PRIMARY_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/DemoteRequest.cc b/src/librbd/mirror/snapshot/DemoteRequest.cc
new file mode 100644
index 000000000..ccaa33c83
--- /dev/null
+++ b/src/librbd/mirror/snapshot/DemoteRequest.cc
@@ -0,0 +1,110 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/DemoteRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::DemoteRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void DemoteRequest<I>::send() {
+  enable_non_primary_feature();
+}
+
+template <typename I>
+void DemoteRequest<I>::enable_non_primary_feature() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  // ensure image is flagged with non-primary feature so that
+  // standard RBD clients cannot write to it.
+  librados::ObjectWriteOperation op;
+  cls_client::set_features(&op, RBD_FEATURE_NON_PRIMARY,
+                           RBD_FEATURE_NON_PRIMARY);
+
+  auto aio_comp = create_rados_callback<
+    DemoteRequest<I>,
+    &DemoteRequest<I>::handle_enable_non_primary_feature>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
+                                          &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_enable_non_primary_feature(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to enable non-primary feature: "
+               << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  create_snapshot();
+}
+
+template <typename I>
+void DemoteRequest<I>::create_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    DemoteRequest<I>, &DemoteRequest<I>::handle_create_snapshot>(this);
+
+  auto req = CreatePrimaryRequest<I>::create(
+    m_image_ctx, m_global_image_id, CEPH_NOSNAP,
+    SNAP_CREATE_FLAG_SKIP_NOTIFY_QUIESCE,
+    (snapshot::CREATE_PRIMARY_FLAG_IGNORE_EMPTY_PEERS |
+     snapshot::CREATE_PRIMARY_FLAG_DEMOTED), nullptr, ctx);
+  req->send();
+}
+
+template <typename I>
+void DemoteRequest<I>::handle_create_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create mirror snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void DemoteRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::DemoteRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/DemoteRequest.h b/src/librbd/mirror/snapshot/DemoteRequest.h
new file mode 100644
index 000000000..63c935645
--- /dev/null
+++ b/src/librbd/mirror/snapshot/DemoteRequest.h
@@ -0,0 +1,76 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_DEMOTE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_DEMOTE_REQUEST_H
+
+#include "include/buffer.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class DemoteRequest {
+public:
+  static DemoteRequest *create(ImageCtxT *image_ctx,
+                               const std::string& global_image_id,
+                               Context *on_finish) {
+    return new DemoteRequest(image_ctx, global_image_id, on_finish);
+  }
+
+  DemoteRequest(ImageCtxT *image_ctx, const std::string& global_image_id,
+                Context *on_finish)
+    : m_image_ctx(image_ctx), m_global_image_id(global_image_id),
+      m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * ENABLE_NON_PRIMARY_FEATURE
+   *    |
+   *    v
+   * CREATE_SNAPSHOT
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  std::string m_global_image_id;
+  Context *m_on_finish;
+
+  void enable_non_primary_feature();
+  void handle_enable_non_primary_feature(int r);
+
+  void create_snapshot();
+  void handle_create_snapshot(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::DemoteRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_DEMOTE_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/GetImageStateRequest.cc b/src/librbd/mirror/snapshot/GetImageStateRequest.cc
new file mode 100644
index 000000000..4692f88cb
--- /dev/null
+++ b/src/librbd/mirror/snapshot/GetImageStateRequest.cc
@@ -0,0 +1,114 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/GetImageStateRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/Types.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::GetImageStateRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void GetImageStateRequest<I>::send() {
+  read_object();
+}
+
+
+template <typename I>
+void GetImageStateRequest<I>::read_object() {
+  CephContext *cct = m_image_ctx->cct;
+
+  auto oid = util::image_state_object_name(m_image_ctx, m_snap_id,
+                                           m_object_index);
+  ldout(cct, 15) << oid << dendl;
+
+  librados::ObjectReadOperation op;
+  m_bl.clear();
+  op.read(0, 0, &m_bl, nullptr);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    GetImageStateRequest<I>,
+    &GetImageStateRequest<I>::handle_read_object>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(oid, comp, &op, nullptr);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void GetImageStateRequest<I>::handle_read_object(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to read image state object: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  auto iter = m_bl.cbegin();
+
+  if (m_object_index == 0) {
+    ImageStateHeader header;
+    try {
+      using ceph::decode;
+      decode(header, iter);
+    } catch (const buffer::error &err) {
+      lderr(cct) << "failed to decode image state object header" << dendl;
+      finish(-EBADMSG);
+      return;
+    }
+    m_object_count = header.object_count;
+  }
+
+  bufferlist bl;
+  bl.substr_of(m_bl, iter.get_off(), m_bl.length() - iter.get_off());
+  m_state_bl.claim_append(bl);
+
+  m_object_index++;
+
+  if (m_object_index >= m_object_count) {
+    finish(0);
+    return;
+  }
+
+  read_object();
+}
+
+template <typename I>
+void GetImageStateRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r == 0) {
+    try {
+      using ceph::decode;
+      decode(*m_image_state, m_state_bl);
+    } catch (const buffer::error &err) {
+      lderr(cct) << "failed to decode image state" << dendl;
+      r = -EBADMSG;
+    }
+  }
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::GetImageStateRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/GetImageStateRequest.h b/src/librbd/mirror/snapshot/GetImageStateRequest.h
new file mode 100644
index 000000000..483e3a228
--- /dev/null
+++ b/src/librbd/mirror/snapshot/GetImageStateRequest.h
@@ -0,0 +1,76 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_GET_IMAGE_STATE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_GET_IMAGE_STATE_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/types.h"
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+struct ImageState;
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class GetImageStateRequest {
+public:
+  static GetImageStateRequest *create(ImageCtxT *image_ctx, uint64_t snap_id,
+                                      ImageState *image_state,
+                                      Context *on_finish) {
+    return new GetImageStateRequest(image_ctx, snap_id, image_state, on_finish);
+  }
+
+  GetImageStateRequest(ImageCtxT *image_ctx, uint64_t snap_id,
+                       ImageState *image_state, Context *on_finish)
+    : m_image_ctx(image_ctx), m_snap_id(snap_id), m_image_state(image_state),
+      m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * READ_OBJECT (repeat for
+   *    |         every object)
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  uint64_t m_snap_id;
+  ImageState *m_image_state;
+  Context *m_on_finish;
+
+  bufferlist m_bl;
+  bufferlist m_state_bl;
+
+  size_t m_object_count = 0;
+  size_t m_object_index = 0;
+
+  void read_object();
+  void handle_read_object(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::GetImageStateRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_GET_IMAGE_STATE_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/ImageMeta.cc b/src/librbd/mirror/snapshot/ImageMeta.cc
new file mode 100644
index 000000000..826899775
--- /dev/null
+++ b/src/librbd/mirror/snapshot/ImageMeta.cc
@@ -0,0 +1,175 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/ImageMeta.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "json_spirit/json_spirit.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/WatchNotifyTypes.h"
+#include "librbd/mirror/snapshot/Utils.h"
+#include "librbd/watcher/Notifier.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::ImageMeta: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_rados_callback;
+using librbd::mirror::snapshot::util::get_image_meta_key;
+
+template <typename I>
+ImageMeta<I>::ImageMeta(I* image_ctx, const std::string& mirror_uuid)
+  : m_image_ctx(image_ctx), m_mirror_uuid(mirror_uuid) {
+}
+
+template <typename I>
+void ImageMeta<I>::load(Context* on_finish) {
+  ldout(m_image_ctx->cct, 15) << "oid=" << m_image_ctx->header_oid << ", "
+                              << "key=" << get_image_meta_key(m_mirror_uuid)
+                              << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::metadata_get_start(&op, get_image_meta_key(m_mirror_uuid));
+
+  m_out_bl.clear();
+  auto ctx = new LambdaContext([this, on_finish](int r) {
+      handle_load(on_finish, r);
+    });
+  auto aio_comp = create_rados_callback(ctx);
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
+                                          &op, &m_out_bl);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void ImageMeta<I>::handle_load(Context* on_finish, int r) {
+  ldout(m_image_ctx->cct, 15) << "r=" << r << dendl;
+
+  std::string data;
+  if (r == 0) {
+    auto it = m_out_bl.cbegin();
+    r = cls_client::metadata_get_finish(&it, &data);
+  }
+
+  if (r == -ENOENT) {
+    ldout(m_image_ctx->cct, 15) << "no snapshot-based mirroring image-meta: "
+                                << cpp_strerror(r) << dendl;
+    on_finish->complete(r);
+    return;
+  } else if (r < 0) {
+    lderr(m_image_ctx->cct) << "failed to load snapshot-based mirroring "
+                            << "image-meta: " << cpp_strerror(r) << dendl;
+    on_finish->complete(r);
+    return;
+  }
+
+  bool json_valid = false;
+  json_spirit::mValue json_root;
+  if (json_spirit::read(data, json_root)) {
+    try {
+      auto& json_obj = json_root.get_obj();
+      resync_requested = json_obj["resync_requested"].get_bool();
+      json_valid = true;
+    } catch (std::runtime_error&) {
+    }
+  }
+
+  if (!json_valid) {
+    lderr(m_image_ctx->cct) << "invalid image-meta JSON received" << dendl;
+    on_finish->complete(-EBADMSG);
+    return;
+  }
+
+  on_finish->complete(0);
+}
+
+template <typename I>
+void ImageMeta<I>::save(Context* on_finish) {
+  ldout(m_image_ctx->cct, 15) << "oid=" << m_image_ctx->header_oid << ", "
+                              << "key=" << get_image_meta_key(m_mirror_uuid)
+                              << dendl;
+
+  // simple implementation for now
+  std::string json = "{\"resync_requested\": " +
+                     std::string(resync_requested ? "true" : "false") + "}";
+
+  bufferlist bl;
+  bl.append(json);
+
+  // avoid using built-in metadata_set operation since that would require
+  // opening the non-primary image in read/write mode which isn't supported
+  librados::ObjectWriteOperation op;
+  cls_client::metadata_set(&op, {{get_image_meta_key(m_mirror_uuid), bl}});
+
+  auto ctx = new LambdaContext([this, on_finish](int r) {
+      handle_save(on_finish, r);
+    });
+  auto aio_comp = create_rados_callback(ctx);
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
+                                          &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void ImageMeta<I>::handle_save(Context* on_finish, int r) {
+  ldout(m_image_ctx->cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_image_ctx->cct) << "failed to save snapshot-based mirroring "
+                            << "image-meta: " << cpp_strerror(r) << dendl;
+    on_finish->complete(r);
+    return;
+  }
+
+  notify_update(on_finish);
+}
+
+template <typename I>
+void ImageMeta<I>::notify_update(Context* on_finish) {
+  ldout(m_image_ctx->cct, 15) << dendl;
+
+  // directly send header notification on image since you cannot
+  // open a non-primary image read/write and therefore cannot re-use
+  // the ImageWatcher to send the notification
+  bufferlist bl;
+  encode(watch_notify::NotifyMessage(new watch_notify::HeaderUpdatePayload()),
+         bl);
+
+  m_out_bl.clear();
+  auto ctx = new LambdaContext([this, on_finish](int r) {
+      handle_notify_update(on_finish, r);
+    });
+  auto aio_comp = create_rados_callback(ctx);
+  int r = m_image_ctx->md_ctx.aio_notify(
+    m_image_ctx->header_oid, aio_comp, bl, watcher::Notifier::NOTIFY_TIMEOUT,
+    &m_out_bl);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void ImageMeta<I>::handle_notify_update(Context* on_finish, int r) {
+  ldout(m_image_ctx->cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(m_image_ctx->cct) << "failed to notify image update: "
+                            << cpp_strerror(r) << dendl;
+  }
+  on_finish->complete(r);
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::ImageMeta<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/ImageMeta.h b/src/librbd/mirror/snapshot/ImageMeta.h
new file mode 100644
index 000000000..5d05f1927
--- /dev/null
+++ b/src/librbd/mirror/snapshot/ImageMeta.h
@@ -0,0 +1,78 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_IMAGE_META_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_IMAGE_META_H
+
+#include "include/rados/librados.hpp"
+#include <string>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT>
+class ImageMeta {
+public:
+  static ImageMeta* create(ImageCtxT* image_ctx,
+                           const std::string& mirror_uuid) {
+    return new ImageMeta(image_ctx, mirror_uuid);
+  }
+
+  ImageMeta(ImageCtxT* image_ctx, const std::string& mirror_uuid);
+
+  void load(Context* on_finish);
+  void save(Context* on_finish);
+
+  bool resync_requested = false;
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *   |
+   *   v
+   * METADATA_GET
+   *   |
+   *   v
+   * <idle>
+   *   |
+   *   v
+   * METADATA_SET
+   *   |
+   *   v
+   * NOTIFY_UPDATE
+   *   |
+   *   v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT* m_image_ctx;
+  std::string m_mirror_uuid;
+
+  bufferlist m_out_bl;
+
+  void handle_load(Context* on_finish, int r);
+
+  void handle_save(Context* on_finish, int r);
+
+  void notify_update(Context* on_finish);
+  void handle_notify_update(Context* on_finish, int r);
+
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::ImageMeta<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_IMAGE_META_H
diff --git a/src/librbd/mirror/snapshot/PromoteRequest.cc b/src/librbd/mirror/snapshot/PromoteRequest.cc
new file mode 100644
index 000000000..9718c299e
--- /dev/null
+++ b/src/librbd/mirror/snapshot/PromoteRequest.cc
@@ -0,0 +1,405 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/PromoteRequest.h"
+#include "common/Timer.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ExclusiveLock.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+#include "librbd/asio/ContextWQ.h"
+#include "librbd/image/ListWatchersRequest.h"
+#include "librbd/mirror/snapshot/CreateNonPrimaryRequest.h"
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::PromoteRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_async_context_callback;
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void PromoteRequest<I>::send() {
+  CephContext *cct = m_image_ctx->cct;
+  bool requires_orphan = false;
+  if (!util::can_create_primary_snapshot(m_image_ctx, false, true,
+                                         &requires_orphan,
+                                         &m_rollback_snap_id)) {
+    lderr(cct) << "cannot promote" << dendl;
+    finish(-EINVAL);
+    return;
+  } else if (m_rollback_snap_id == CEPH_NOSNAP && !requires_orphan) {
+    create_promote_snapshot();
+    return;
+  }
+
+  ldout(cct, 15) << "requires_orphan=" << requires_orphan << ", "
+                 << "rollback_snap_id=" << m_rollback_snap_id << dendl;
+  create_orphan_snapshot();
+}
+
+template <typename I>
+void PromoteRequest<I>::create_orphan_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>,
+    &PromoteRequest<I>::handle_create_orphan_snapshot>(this);
+
+  auto req = CreateNonPrimaryRequest<I>::create(
+    m_image_ctx, false, "", CEPH_NOSNAP, {}, {}, nullptr, ctx);
+  req->send();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_create_orphan_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create orphan snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  list_watchers();
+}
+
+template <typename I>
+void PromoteRequest<I>::list_watchers() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>,
+    &PromoteRequest<I>::handle_list_watchers>(this);
+
+  m_watchers.clear();
+  auto flags = librbd::image::LIST_WATCHERS_FILTER_OUT_MY_INSTANCE |
+               librbd::image::LIST_WATCHERS_MIRROR_INSTANCES_ONLY;
+  auto req = librbd::image::ListWatchersRequest<I>::create(
+    *m_image_ctx, flags, &m_watchers, ctx);
+  req->send();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_list_watchers(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to list watchers: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_watchers.empty()) {
+    acquire_exclusive_lock();
+    return;
+  }
+
+  wait_update_notify();
+}
+
+template <typename I>
+void PromoteRequest<I>::wait_update_notify() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  ImageCtx::get_timer_instance(cct, &m_timer, &m_timer_lock);
+
+  std::lock_guard timer_lock{*m_timer_lock};
+
+  m_scheduler_ticks = 5;
+
+  int r = m_image_ctx->state->register_update_watcher(&m_update_watch_ctx,
+                                                      &m_update_watcher_handle);
+  if (r < 0) {
+    lderr(cct) << "failed to register update watcher: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  scheduler_unregister_update_watcher();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_update_notify() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  std::lock_guard timer_lock{*m_timer_lock};
+  m_scheduler_ticks = 0;
+}
+
+template <typename I>
+void PromoteRequest<I>::scheduler_unregister_update_watcher() {
+  ceph_assert(ceph_mutex_is_locked(*m_timer_lock));
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "scheduler_ticks=" << m_scheduler_ticks << dendl;
+
+  if (m_scheduler_ticks > 0) {
+    m_scheduler_ticks--;
+    m_timer->add_event_after(1, new LambdaContext([this](int) {
+        scheduler_unregister_update_watcher();
+      }));
+    return;
+  }
+
+  m_image_ctx->op_work_queue->queue(new LambdaContext([this](int) {
+      unregister_update_watcher();
+    }), 0);
+}
+
+template <typename I>
+void PromoteRequest<I>::unregister_update_watcher() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>,
+    &PromoteRequest<I>::handle_unregister_update_watcher>(this);
+
+  m_image_ctx->state->unregister_update_watcher(m_update_watcher_handle, ctx);
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_unregister_update_watcher(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to unregister update watcher: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  list_watchers();
+}
+
+template <typename I>
+void PromoteRequest<I>::acquire_exclusive_lock() {
+  {
+    std::unique_lock locker{m_image_ctx->owner_lock};
+    if (m_image_ctx->exclusive_lock != nullptr &&
+        !m_image_ctx->exclusive_lock->is_lock_owner()) {
+      CephContext *cct = m_image_ctx->cct;
+      ldout(cct, 15) << dendl;
+
+      m_lock_acquired = true;
+      m_image_ctx->exclusive_lock->block_requests(0);
+
+      auto ctx = create_context_callback<
+        PromoteRequest<I>,
+        &PromoteRequest<I>::handle_acquire_exclusive_lock>(this);
+
+      m_image_ctx->exclusive_lock->acquire_lock(ctx);
+      return;
+    }
+  }
+
+  rollback();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_acquire_exclusive_lock(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to acquire exclusive lock: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  } else {
+    std::unique_lock locker{m_image_ctx->owner_lock};
+    if (m_image_ctx->exclusive_lock != nullptr &&
+        !m_image_ctx->exclusive_lock->is_lock_owner()) {
+      lderr(cct) << "failed to acquire exclusive lock" << dendl;
+      r = m_image_ctx->exclusive_lock->get_unlocked_op_error();
+      locker.unlock();
+      finish(r);
+      return;
+    }
+  }
+
+  rollback();
+}
+
+template <typename I>
+void PromoteRequest<I>::rollback() {
+  if (m_rollback_snap_id == CEPH_NOSNAP) {
+    create_promote_snapshot();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  std::shared_lock owner_locker{m_image_ctx->owner_lock};
+  std::shared_lock image_locker{m_image_ctx->image_lock};
+
+  auto info = m_image_ctx->get_snap_info(m_rollback_snap_id);
+  ceph_assert(info != nullptr);
+  auto snap_namespace = info->snap_namespace;
+  auto snap_name = info->name;
+
+  image_locker.unlock();
+
+  auto ctx = create_async_context_callback(
+    *m_image_ctx, create_context_callback<
+      PromoteRequest<I>, &PromoteRequest<I>::handle_rollback>(this));
+
+  m_image_ctx->operations->execute_snap_rollback(snap_namespace, snap_name,
+                                                 m_progress_ctx, ctx);
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_rollback(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to rollback: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  create_promote_snapshot();
+}
+
+template <typename I>
+void PromoteRequest<I>::create_promote_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    PromoteRequest<I>,
+    &PromoteRequest<I>::handle_create_promote_snapshot>(this);
+
+  auto req = CreatePrimaryRequest<I>::create(
+    m_image_ctx, m_global_image_id, CEPH_NOSNAP,
+    SNAP_CREATE_FLAG_SKIP_NOTIFY_QUIESCE,
+    (snapshot::CREATE_PRIMARY_FLAG_IGNORE_EMPTY_PEERS |
+     snapshot::CREATE_PRIMARY_FLAG_FORCE), nullptr, ctx);
+  req->send();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_create_promote_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create promote snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  disable_non_primary_feature();
+}
+
+template <typename I>
+void PromoteRequest<I>::disable_non_primary_feature() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << dendl;
+
+  // remove the non-primary feature flag so that the image can be
+  // R/W by standard RBD clients
+  librados::ObjectWriteOperation op;
+  cls_client::set_features(&op, 0U, RBD_FEATURE_NON_PRIMARY);
+
+  auto aio_comp = create_rados_callback<
+    PromoteRequest<I>,
+    &PromoteRequest<I>::handle_disable_non_primary_feature>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
+                                          &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_disable_non_primary_feature(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to disable non-primary feature: "
+               << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  release_exclusive_lock();
+}
+
+template <typename I>
+void PromoteRequest<I>::release_exclusive_lock() {
+  if (m_lock_acquired) {
+    std::unique_lock locker{m_image_ctx->owner_lock};
+    if (m_image_ctx->exclusive_lock != nullptr) {
+      CephContext *cct = m_image_ctx->cct;
+      ldout(cct, 15) << dendl;
+
+      m_image_ctx->exclusive_lock->unblock_requests();
+
+      auto ctx = create_context_callback<
+        PromoteRequest<I>,
+        &PromoteRequest<I>::handle_release_exclusive_lock>(this);
+
+      m_image_ctx->exclusive_lock->release_lock(ctx);
+      return;
+    }
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_release_exclusive_lock(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to release exclusive lock: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void PromoteRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::PromoteRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/PromoteRequest.h b/src/librbd/mirror/snapshot/PromoteRequest.h
new file mode 100644
index 000000000..1d9a862a0
--- /dev/null
+++ b/src/librbd/mirror/snapshot/PromoteRequest.h
@@ -0,0 +1,151 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_PROMOTE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_PROMOTE_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/rbd/librbd.hpp"
+#include "common/ceph_mutex.h"
+#include "common/Timer.h"
+#include "librbd/internal.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class PromoteRequest {
+public:
+  static PromoteRequest *create(ImageCtxT *image_ctx,
+                                const std::string& global_image_id,
+                                Context *on_finish) {
+    return new PromoteRequest(image_ctx, global_image_id, on_finish);
+  }
+
+  PromoteRequest(ImageCtxT *image_ctx, const std::string& global_image_id,
+                 Context *on_finish)
+    : m_image_ctx(image_ctx), m_global_image_id(global_image_id),
+      m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    |          (can promote)
+   *    |\----------------------------------------\
+   *    |                                         |
+   *    |                                         |
+   *    v (skip if not needed)                    |
+   * CREATE_ORPHAN_SNAPSHOT                       |
+   *    |                                         |
+   *    |     /-- UNREGISTER_UPDATE_WATCHER <-\   |
+   *    v     v                               |   |
+   * LIST_WATCHERS ----> WAIT_UPDATE_NOTIFY --/   |
+   *    |                                         |
+   *    | (no watchers)                           |
+   *    v                                         |
+   * ACQUIRE_EXCLUSIVE_LOCK                       |
+   *    |  (skip if not needed)                   |
+   *    v                                         |
+   * ROLLBACK                                     |
+   *    |                                         |
+   *    v                                         |
+   * CREATE_PROMOTE_SNAPSHOT <--------------------/
+   *    |
+   *    v
+   * DISABLE_NON_PRIMARY_FEATURE
+   *    |
+   *    v
+   * RELEASE_EXCLUSIVE_LOCK (skip if not needed)
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  std::string m_global_image_id;
+  Context *m_on_finish;
+
+  uint64_t m_rollback_snap_id = CEPH_NOSNAP;
+  bool m_lock_acquired = false;
+  NoOpProgressContext m_progress_ctx;
+
+  class UpdateWatchCtx : public librbd::UpdateWatchCtx {
+  public:
+    UpdateWatchCtx(PromoteRequest *promote_request)
+      : promote_request(promote_request) {
+    }
+
+    void handle_notify() {
+      promote_request->handle_update_notify();
+    }
+
+  private:
+    PromoteRequest *promote_request;
+
+  } m_update_watch_ctx = {this};
+
+  std::list<obj_watch_t> m_watchers;
+  uint64_t m_update_watcher_handle = 0;
+  uint64_t m_scheduler_ticks = 0;
+  SafeTimer *m_timer = nullptr;
+  ceph::mutex *m_timer_lock = nullptr;
+
+  void refresh_image();
+  void handle_refresh_image(int r);
+
+  void create_orphan_snapshot();
+  void handle_create_orphan_snapshot(int r);
+
+  void list_watchers();
+  void handle_list_watchers(int r);
+
+  void wait_update_notify();
+  void handle_update_notify();
+  void scheduler_unregister_update_watcher();
+
+  void unregister_update_watcher();
+  void handle_unregister_update_watcher(int r);
+
+  void acquire_exclusive_lock();
+  void handle_acquire_exclusive_lock(int r);
+
+  void rollback();
+  void handle_rollback(int r);
+
+  void create_promote_snapshot();
+  void handle_create_promote_snapshot(int r);
+
+  void disable_non_primary_feature();
+  void handle_disable_non_primary_feature(int r);
+
+  void release_exclusive_lock();
+  void handle_release_exclusive_lock(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::PromoteRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_PROMOTE_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/RemoveImageStateRequest.cc b/src/librbd/mirror/snapshot/RemoveImageStateRequest.cc
new file mode 100644
index 000000000..204e0489a
--- /dev/null
+++ b/src/librbd/mirror/snapshot/RemoveImageStateRequest.cc
@@ -0,0 +1,131 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/RemoveImageStateRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/Types.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::RemoveImageStateRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void RemoveImageStateRequest<I>::send() {
+  get_object_count();
+}
+
+
+template <typename I>
+void RemoveImageStateRequest<I>::get_object_count() {
+  CephContext *cct = m_image_ctx->cct;
+
+  auto oid = util::image_state_object_name(m_image_ctx, m_snap_id, 0);
+  ldout(cct, 15) << oid << dendl;
+
+  librados::ObjectReadOperation op;
+  op.read(0, 0, &m_bl, nullptr);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    RemoveImageStateRequest<I>,
+    &RemoveImageStateRequest<I>::handle_get_object_count>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(oid, comp, &op, nullptr);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void RemoveImageStateRequest<I>::handle_get_object_count(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to read image state object: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  ImageStateHeader header(1);
+  auto iter = m_bl.cbegin();
+  try {
+    using ceph::decode;
+    
+    decode(header, iter);
+  } catch (const buffer::error &err) {
+    lderr(cct) << "failed to decode image state object header" << dendl;
+    // still try to remove it
+  }
+
+  m_object_count = header.object_count > 0 ? header.object_count : 1;
+
+  remove_object();
+}
+
+template <typename I>
+void RemoveImageStateRequest<I>::remove_object() {
+  CephContext *cct = m_image_ctx->cct;
+
+  ceph_assert(m_object_count > 0);
+  m_object_count--;
+
+  auto oid = util::image_state_object_name(m_image_ctx, m_snap_id,
+                                           m_object_count);
+  ldout(cct, 15) << oid << dendl;
+
+  librados::ObjectWriteOperation op;
+  op.remove();
+
+  librados::AioCompletion *comp = create_rados_callback<
+    RemoveImageStateRequest<I>,
+    &RemoveImageStateRequest<I>::handle_remove_object>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(oid, comp, &op);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void RemoveImageStateRequest<I>::handle_remove_object(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(cct) << "failed to remove image state object: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_object_count == 0) {
+    finish(0);
+    return;
+  }
+
+  remove_object();
+}
+
+template <typename I>
+void RemoveImageStateRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::RemoveImageStateRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/RemoveImageStateRequest.h b/src/librbd/mirror/snapshot/RemoveImageStateRequest.h
new file mode 100644
index 000000000..be7dad8e0
--- /dev/null
+++ b/src/librbd/mirror/snapshot/RemoveImageStateRequest.h
@@ -0,0 +1,75 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_REMOVE_IMAGE_STATE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_REMOVE_IMAGE_STATE_REQUEST_H
+
+#include "include/buffer.h"
+#include "include/types.h"
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class RemoveImageStateRequest {
+public:
+  static RemoveImageStateRequest *create(ImageCtxT *image_ctx, uint64_t snap_id,
+                                         Context *on_finish) {
+      return new RemoveImageStateRequest(image_ctx, snap_id, on_finish);
+  }
+
+  RemoveImageStateRequest(ImageCtxT *image_ctx, uint64_t snap_id,
+                          Context *on_finish)
+    : m_image_ctx(image_ctx), m_snap_id(snap_id), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_OBJECT_COUNT
+   *    |
+   *    v
+   * REMOVE_OBJECT (repeat for
+   *    |           every object)
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  uint64_t m_snap_id;
+  Context *m_on_finish;
+
+  bufferlist m_bl;
+
+  size_t m_object_count = 0;
+
+  void get_object_count();
+  void handle_get_object_count(int r);
+
+  void remove_object();
+  void handle_remove_object(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::RemoveImageStateRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_REMOVE_IMAGE_STATE_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/SetImageStateRequest.cc b/src/librbd/mirror/snapshot/SetImageStateRequest.cc
new file mode 100644
index 000000000..9fcee0322
--- /dev/null
+++ b/src/librbd/mirror/snapshot/SetImageStateRequest.cc
@@ -0,0 +1,235 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/SetImageStateRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/image/GetMetadataRequest.h"
+#include "librbd/mirror/snapshot/WriteImageStateRequest.h"
+
+#include <boost/algorithm/string/predicate.hpp>
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror_snapshot::SetImageStateRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void SetImageStateRequest<I>::send() {
+  get_name();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::get_name() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::dir_get_name_start(&op, m_image_ctx->id);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    SetImageStateRequest<I>,
+    &SetImageStateRequest<I>::handle_get_name>(this);
+  m_bl.clear();
+  int r = m_image_ctx->md_ctx.aio_operate(RBD_DIRECTORY, comp, &op, &m_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::handle_get_name(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r == 0) {
+    auto it = m_bl.cbegin();
+    r = cls_client::dir_get_name_finish(&it, &m_image_state.name);
+  }
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve image name: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  ldout(cct, 15) << "name=" << m_image_state.name << dendl;
+
+  get_snap_limit();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::get_snap_limit() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::snapshot_get_limit_start(&op);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    SetImageStateRequest<I>,
+    &SetImageStateRequest<I>::handle_get_snap_limit>(this);
+  m_bl.clear();
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, comp, &op,
+                                          &m_bl);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::handle_get_snap_limit(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r == 0) {
+    auto it = m_bl.cbegin();
+    r = cls_client::snapshot_get_limit_finish(&it, &m_image_state.snap_limit);
+  }
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve snapshot limit: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  ldout(cct, 15) << "snap_limit=" << m_image_state.snap_limit << dendl;
+
+  get_metadata();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::get_metadata() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+     SetImageStateRequest<I>,
+     &SetImageStateRequest<I>::handle_get_metadata>(this);
+  auto req = image::GetMetadataRequest<I>::create(
+    m_image_ctx->md_ctx, m_image_ctx->header_oid, true, "", "", 0,
+    &m_image_state.metadata, ctx);
+  req->send();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::handle_get_metadata(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to retrieve metadata: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+
+    m_image_state.features =
+      m_image_ctx->features & ~RBD_FEATURES_IMPLICIT_ENABLE;
+
+    for (auto &[snap_id, snap_info] : m_image_ctx->snap_info) {
+      auto type = cls::rbd::get_snap_namespace_type(snap_info.snap_namespace);
+      if (type != cls::rbd::SNAPSHOT_NAMESPACE_TYPE_USER) {
+        // only replicate user snapshots -- trash snapshots will be
+        // replicated by an implicit delete if required
+        continue;
+      }
+      m_image_state.snapshots[snap_id] = {snap_info.snap_namespace,
+                                          snap_info.name,
+                                          snap_info.protection_status};
+    }
+  }
+
+  write_image_state();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::write_image_state() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    SetImageStateRequest<I>,
+    &SetImageStateRequest<I>::handle_write_image_state>(this);
+
+  auto req = WriteImageStateRequest<I>::create(m_image_ctx, m_snap_id,
+                                               m_image_state, ctx);
+  req->send();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::handle_write_image_state(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to write image state: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  update_primary_snapshot();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::update_primary_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  librados::ObjectWriteOperation op;
+  librbd::cls_client::mirror_image_snapshot_set_copy_progress(
+    &op, m_snap_id, true, 0);
+
+  auto aio_comp = create_rados_callback<
+    SetImageStateRequest<I>,
+    &SetImageStateRequest<I>::handle_update_primary_snapshot>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
+                                          &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void SetImageStateRequest<I>::handle_update_primary_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to update primary snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void SetImageStateRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::SetImageStateRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/SetImageStateRequest.h b/src/librbd/mirror/snapshot/SetImageStateRequest.h
new file mode 100644
index 000000000..fd2815494
--- /dev/null
+++ b/src/librbd/mirror/snapshot/SetImageStateRequest.h
@@ -0,0 +1,96 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_SET_IMAGE_STATE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_SET_IMAGE_STATE_REQUEST_H
+
+#include "librbd/mirror/snapshot/Types.h"
+
+#include <map>
+#include <string>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class SetImageStateRequest {
+public:
+  static SetImageStateRequest *create(ImageCtxT *image_ctx, uint64_t snap_id,
+                                      Context *on_finish) {
+    return new SetImageStateRequest(image_ctx, snap_id, on_finish);
+  }
+
+  SetImageStateRequest(ImageCtxT *image_ctx, uint64_t snap_id,
+                       Context *on_finish)
+    : m_image_ctx(image_ctx), m_snap_id(snap_id), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_NAME
+   *    |
+   *    v
+   * GET_SNAP_LIMIT
+   *    |
+   *    v
+   * GET_METADATA
+   *    |
+   *    v
+   * WRITE_IMAGE_STATE
+   *    |
+   *    v
+   * UPDATE_PRIMARY_SNAPSHOT
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  uint64_t m_snap_id;
+  Context *m_on_finish;
+
+  ImageState m_image_state;
+
+  bufferlist m_bl;
+  bufferlist m_state_bl;
+
+  void get_name();
+  void handle_get_name(int r);
+
+  void get_snap_limit();
+  void handle_get_snap_limit(int r);
+
+  void get_metadata();
+  void handle_get_metadata(int r);
+
+  void write_image_state();
+  void handle_write_image_state(int r);
+
+  void update_primary_snapshot();
+  void handle_update_primary_snapshot(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::SetImageStateRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_SET_IMAGE_STATE_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/Types.cc b/src/librbd/mirror/snapshot/Types.cc
new file mode 100644
index 000000000..866b4c3e2
--- /dev/null
+++ b/src/librbd/mirror/snapshot/Types.cc
@@ -0,0 +1,109 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "common/Formatter.h"
+#include "include/encoding.h"
+#include "include/stringify.h"
+#include "librbd/mirror/snapshot/Types.h"
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+void ImageStateHeader::encode(bufferlist& bl) const {
+  ENCODE_START(1, 1, bl);
+  encode(object_count, bl);
+  ENCODE_FINISH(bl);
+}
+
+void ImageStateHeader::decode(bufferlist::const_iterator& bl) {
+  DECODE_START(1, bl);
+  decode(object_count, bl);
+  DECODE_FINISH(bl);
+}
+
+void SnapState::encode(bufferlist& bl) const {
+  ENCODE_START(1, 1, bl);
+  encode(snap_namespace, bl);
+  encode(name, bl);
+  encode(protection_status, bl);
+  ENCODE_FINISH(bl);
+}
+
+void SnapState::decode(bufferlist::const_iterator& bl) {
+  DECODE_START(1, bl);
+  decode(snap_namespace, bl);
+  decode(name, bl);
+  decode(protection_status, bl);
+  DECODE_FINISH(bl);
+}
+
+void SnapState::dump(Formatter *f) const {
+  f->open_object_section("namespace");
+  snap_namespace.dump(f);
+  f->close_section();
+  f->dump_string("name", name);
+  f->dump_unsigned("protection_status", protection_status);
+}
+
+std::ostream& operator<<(std::ostream& os, const SnapState& snap_state) {
+  os << "["
+     << "namespace=" << snap_state.snap_namespace << ", "
+     << "name=" << snap_state.name << ", "
+     << "protection=" << static_cast<int>(snap_state.protection_status)
+     << "]";
+  return os;
+}
+
+void ImageState::encode(bufferlist& bl) const {
+  ENCODE_START(1, 1, bl);
+  encode(name, bl);
+  encode(features, bl);
+  encode(snap_limit, bl);
+  encode(snapshots, bl);
+  encode(metadata, bl);
+  ENCODE_FINISH(bl);
+}
+
+void ImageState::decode(bufferlist::const_iterator& bl) {
+  DECODE_START(1, bl);
+  decode(name, bl);
+  decode(features, bl);
+  decode(snap_limit, bl);
+  decode(snapshots, bl);
+  decode(metadata, bl);
+  DECODE_FINISH(bl);
+}
+
+void ImageState::dump(Formatter *f) const {
+  f->dump_string("name", name);
+  f->dump_unsigned("features", features);
+  f->dump_unsigned("snap_limit", snap_limit);
+  f->open_array_section("snapshots");
+  for (auto &[id, snap_state] : snapshots) {
+    f->open_object_section(stringify(id).c_str());
+    snap_state.dump(f);
+    f->close_section(); // snap_state
+  }
+  f->close_section(); // snapshots
+  f->open_object_section("metadata");
+  for (auto &it : metadata) {
+    f->dump_stream(it.first.c_str()) << it.second;
+  }
+  f->close_section(); // metadata
+}
+
+std::ostream& operator<<(std::ostream& os, const ImageState& image_state) {
+  os << "["
+     << "name=" << image_state.name << ", "
+     << "features=" << image_state.features << ", "
+     << "snap_limit=" << image_state.snap_limit << ", "
+     << "snaps=" << image_state.snapshots << ", "
+     << "metadata_count=" << image_state.metadata.size()
+     << "]";
+  return os;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
diff --git a/src/librbd/mirror/snapshot/Types.h b/src/librbd/mirror/snapshot/Types.h
new file mode 100644
index 000000000..79947a5f8
--- /dev/null
+++ b/src/librbd/mirror/snapshot/Types.h
@@ -0,0 +1,122 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_TYPES_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_TYPES_H
+
+#include "cls/rbd/cls_rbd_types.h"
+#include "include/buffer.h"
+#include "include/types.h"
+
+#include <map>
+#include <string>
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+enum CreatePrimaryFlags {
+  CREATE_PRIMARY_FLAG_IGNORE_EMPTY_PEERS = (1 << 0),
+  CREATE_PRIMARY_FLAG_DEMOTED            = (1 << 1),
+  CREATE_PRIMARY_FLAG_FORCE              = (1 << 2)
+};
+
+struct ImageStateHeader {
+  uint32_t object_count = 0;
+
+  ImageStateHeader() {
+  }
+  ImageStateHeader(uint32_t object_count) : object_count(object_count) {
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(bufferlist::const_iterator &it);
+};
+
+WRITE_CLASS_ENCODER(ImageStateHeader);
+
+struct SnapState {
+  cls::rbd::SnapshotNamespace snap_namespace;
+  std::string name;
+  uint8_t protection_status = 0;
+
+  SnapState() {
+  }
+  SnapState(const cls::rbd::SnapshotNamespace &snap_namespace,
+            const std::string &name, uint8_t protection_status)
+    : snap_namespace(snap_namespace), name(name),
+      protection_status(protection_status) {
+  }
+
+  bool operator==(const SnapState& rhs) const {
+    return snap_namespace == rhs.snap_namespace &&
+           name == rhs.name && protection_status == rhs.protection_status;
+  }
+
+  bool operator<(const SnapState& rhs) const {
+    if (snap_namespace != rhs.snap_namespace) {
+      return snap_namespace < rhs.snap_namespace;
+    }
+    if (name != rhs.name) {
+      return name < rhs.name;
+    }
+    return protection_status < rhs.protection_status;
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(bufferlist::const_iterator &it);
+  void dump(Formatter *f) const;
+};
+
+std::ostream& operator<<(std::ostream& os, const SnapState& snap_state);
+
+WRITE_CLASS_ENCODER(SnapState);
+
+struct ImageState {
+  std::string name;
+  uint64_t features = 0;
+  uint64_t snap_limit = 0;
+  std::map<uint64_t, SnapState> snapshots;
+  std::map<std::string, bufferlist> metadata;
+
+  ImageState() {
+  }
+  ImageState(const std::string &name, uint64_t features, uint64_t snap_limit,
+             const std::map<uint64_t, SnapState> &snapshots,
+             const std::map<std::string, bufferlist> &metadata)
+    : name(name), features(features), snap_limit(snap_limit),
+      snapshots(snapshots), metadata(metadata) {
+  }
+
+  bool operator==(const ImageState& rhs) const {
+    return name == rhs.name && features == rhs.features &&
+           snap_limit == rhs.snap_limit && snapshots == rhs.snapshots;
+  }
+
+  bool operator<(const ImageState& rhs) const {
+    if (name != rhs.name) {
+      return name < rhs.name;
+    }
+    if (features != rhs.features) {
+      return features < rhs.features;
+    }
+    if (snap_limit != rhs.snap_limit) {
+      return snap_limit < rhs.snap_limit;
+    }
+    return snapshots < rhs.snapshots;
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(bufferlist::const_iterator &it);
+  void dump(Formatter *f) const;
+};
+
+std::ostream& operator<<(std::ostream& os, const ImageState& image_state);
+
+WRITE_CLASS_ENCODER(ImageState);
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_TYPES_H
diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
new file mode 100644
index 000000000..6e1884249
--- /dev/null
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
@@ -0,0 +1,235 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/UnlinkPeerRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "librbd/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::UnlinkPeerRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void UnlinkPeerRequest<I>::send() {
+  if (!m_image_ctx->state->is_refresh_required()) {
+    unlink_peer();
+    return;
+  }
+
+  refresh_image();
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::refresh_image() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    UnlinkPeerRequest<I>, &UnlinkPeerRequest<I>::handle_refresh_image>(this);
+  m_image_ctx->state->refresh(ctx);
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::handle_refresh_image(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to refresh image: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  unlink_peer();
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::unlink_peer() {
+  CephContext *cct = m_image_ctx->cct;
+
+  m_image_ctx->image_lock.lock_shared();
+  int r = -ENOENT;
+  cls::rbd::MirrorSnapshotNamespace* mirror_ns = nullptr;
+  m_newer_mirror_snapshots = false;
+  for (auto snap_it = m_image_ctx->snap_info.find(m_snap_id);
+       snap_it != m_image_ctx->snap_info.end(); ++snap_it) {
+    if (snap_it->first == m_snap_id) {
+      r = 0;
+      mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+        &snap_it->second.snap_namespace);
+    } else if (boost::get<cls::rbd::MirrorSnapshotNamespace>(
+                 &snap_it->second.snap_namespace) != nullptr) {
+      ldout(cct, 15) << "located newer mirror snapshot" << dendl;
+      m_newer_mirror_snapshots = true;
+      break;
+    }
+  }
+
+  if (r == -ENOENT) {
+    ldout(cct, 15) << "missing snapshot: snap_id=" << m_snap_id << dendl;
+    m_image_ctx->image_lock.unlock_shared();
+    finish(r);
+    return;
+  }
+
+  if (mirror_ns == nullptr) {
+    lderr(cct) << "not mirror snapshot (snap_id=" << m_snap_id << ")" << dendl;
+    m_image_ctx->image_lock.unlock_shared();
+    finish(-EINVAL);
+    return;
+  }
+
+  // if there is or will be no more peers in the mirror snapshot and we have
+  // a more recent mirror snapshot, remove the older one
+  if ((mirror_ns->mirror_peer_uuids.count(m_mirror_peer_uuid) == 0) ||
+      (mirror_ns->mirror_peer_uuids.size() <= 1U && m_newer_mirror_snapshots)) {
+    m_image_ctx->image_lock.unlock_shared();
+    remove_snapshot();
+    return;
+  }
+  m_image_ctx->image_lock.unlock_shared();
+
+  ldout(cct, 15) << "snap_id=" << m_snap_id << ", "
+                 << "mirror_peer_uuid=" << m_mirror_peer_uuid << dendl;
+  librados::ObjectWriteOperation op;
+  librbd::cls_client::mirror_image_snapshot_unlink_peer(&op, m_snap_id,
+                                                        m_mirror_peer_uuid);
+  auto aio_comp = create_rados_callback<
+    UnlinkPeerRequest<I>, &UnlinkPeerRequest<I>::handle_unlink_peer>(this);
+  r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp, &op);
+  ceph_assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::handle_unlink_peer(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r == -ERESTART || r == -ENOENT) {
+    refresh_image();
+    return;
+  }
+
+  if (r < 0) {
+    lderr(cct) << "failed to unlink peer: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  notify_update();
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::notify_update() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  auto ctx = create_context_callback<
+    UnlinkPeerRequest<I>, &UnlinkPeerRequest<I>::handle_notify_update>(this);
+  m_image_ctx->notify_update(ctx);
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::handle_notify_update(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r == -ENOENT || r == -ETIMEDOUT) {
+    // non-fatel errors
+    lderr(cct) << "failed to notify update: " << cpp_strerror(r) << dendl;
+  } else if (r < 0) {
+    lderr(cct) << "failed to notify update: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  refresh_image();
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::remove_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << dendl;
+
+  cls::rbd::SnapshotNamespace snap_namespace;
+  std::string snap_name;
+  int r = 0;
+  {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+
+    auto snap_info = m_image_ctx->get_snap_info(m_snap_id);
+    if (!snap_info) {
+      r = -ENOENT;
+    } else {
+      snap_namespace = snap_info->snap_namespace;
+      snap_name = snap_info->name;
+    }
+  }
+
+  if (r == -ENOENT) {
+    ldout(cct, 15) << "failed to locate snapshot " << m_snap_id << dendl;
+    finish(0);
+    return;
+  }
+
+  auto info = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+    snap_namespace);
+
+  info.mirror_peer_uuids.erase(m_mirror_peer_uuid);
+  if (!info.mirror_peer_uuids.empty() || !m_newer_mirror_snapshots) {
+    ldout(cct, 15) << "skipping removal of snapshot: "
+                   << "snap_id=" << m_snap_id << ": "
+                   << "mirror_peer_uuid=" << m_mirror_peer_uuid << ", "
+                   << "mirror_peer_uuids=" << info.mirror_peer_uuids << dendl;
+    finish(0);
+    return;
+  }
+
+  auto ctx = create_context_callback<
+    UnlinkPeerRequest<I>, &UnlinkPeerRequest<I>::handle_remove_snapshot>(this);
+  m_image_ctx->operations->snap_remove(snap_namespace, snap_name, ctx);
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::handle_remove_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0 && r != -ENOENT) {
+    lderr(cct) << "failed to remove snapshot: " << cpp_strerror(r) << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void UnlinkPeerRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  auto on_finish = m_on_finish;
+  delete this;
+  on_finish->complete(r);
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::UnlinkPeerRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.h b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
new file mode 100644
index 000000000..9ef47269d
--- /dev/null
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
@@ -0,0 +1,95 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_UNLINK_PEER_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_UNLINK_PEER_REQUEST_H
+
+#include "include/buffer.h"
+
+#include <string>
+#include <set>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class UnlinkPeerRequest {
+public:
+  static UnlinkPeerRequest *create(ImageCtxT *image_ctx, uint64_t snap_id,
+                                   const std::string &mirror_peer_uuid,
+                                   Context *on_finish) {
+    return new UnlinkPeerRequest(image_ctx, snap_id, mirror_peer_uuid,
+                                 on_finish);
+  }
+
+  UnlinkPeerRequest(ImageCtxT *image_ctx, uint64_t snap_id,
+                    const std::string &mirror_peer_uuid, Context *on_finish)
+    : m_image_ctx(image_ctx), m_snap_id(snap_id),
+      m_mirror_peer_uuid(mirror_peer_uuid), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /*
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * REFRESH_IMAGE <--------------------------\
+   *    |                     ^ (not found    |
+   *    |                     *  or last)     |
+   *    |                     *               |
+   *    |\---------------> UNLINK_PEER --> NOTIFY_UPDATE
+   *    |   (not last peer or
+   *    |    no newer mirror
+   *    |    snap exists)
+   *    |
+   *    |\---------------> REMOVE_SNAPSHOT
+   *    |   (last peer and    |
+   *    |    newer mirror     |
+   *    |    snap exists)     |
+   *    |                     |
+   *    |(peer not found)     |
+   *    v                     |
+   * <finish> <---------------/
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  uint64_t m_snap_id;
+  std::string m_mirror_peer_uuid;
+  Context *m_on_finish;
+
+  bool m_newer_mirror_snapshots = false;
+
+  void refresh_image();
+  void handle_refresh_image(int r);
+
+  void unlink_peer();
+  void handle_unlink_peer(int r);
+
+  void notify_update();
+  void handle_notify_update(int r);
+
+  void remove_snapshot();
+  void handle_remove_snapshot(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::UnlinkPeerRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_UNLINK_PEER_REQUEST_H
diff --git a/src/librbd/mirror/snapshot/Utils.cc b/src/librbd/mirror/snapshot/Utils.cc
new file mode 100644
index 000000000..ecf884b54
--- /dev/null
+++ b/src/librbd/mirror/snapshot/Utils.cc
@@ -0,0 +1,186 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "common/dout.h"
+#include "common/errno.h"
+#include "include/stringify.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::util: " \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+namespace util {
+
+namespace {
+
+const std::string IMAGE_STATE_OBJECT_PREFIX = "rbd_mirror_snapshot.";
+
+bool get_rollback_snap_id(
+    std::map<librados::snap_t, SnapInfo>::reverse_iterator it,
+    std::map<librados::snap_t, SnapInfo>::reverse_iterator end,
+    uint64_t *rollback_snap_id) {
+
+  for (; it != end; it++) {
+    auto mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (mirror_ns->state != cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY) {
+      break;
+    }
+    if (mirror_ns->complete) {
+      break;
+    }
+  }
+
+  if (it != end) {
+    *rollback_snap_id = it->first;
+    return true;
+  }
+
+  return false;
+}
+
+} // anonymous namespace
+
+std::string get_image_meta_key(const std::string& mirror_uuid) {
+  return ".rbd_mirror." + mirror_uuid;
+}
+
+template <typename I>
+bool can_create_primary_snapshot(I *image_ctx, bool demoted, bool force,
+                                 bool* requires_orphan,
+                                 uint64_t *rollback_snap_id) {
+  CephContext *cct = image_ctx->cct;
+
+  if (requires_orphan != nullptr) {
+    *requires_orphan = false;
+  }
+  if (rollback_snap_id) {
+    *rollback_snap_id = CEPH_NOSNAP;
+  }
+
+  std::shared_lock image_locker{image_ctx->image_lock};
+
+  for (auto it = image_ctx->snap_info.rbegin();
+       it != image_ctx->snap_info.rend(); it++) {
+    auto mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (mirror_ns == nullptr) {
+      continue;
+    }
+    ldout(cct, 20) << "previous snapshot snap_id=" << it->first << " "
+                   << *mirror_ns << dendl;
+    if (mirror_ns->is_demoted() && !force) {
+      lderr(cct) << "trying to create primary snapshot without force "
+                 << "when previous primary snapshot is demoted"
+                 << dendl;
+      return false;
+    }
+
+    if (mirror_ns->state == cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY) {
+      if (!force) {
+        lderr(cct) << "trying to create primary snapshot without force "
+                   << "when previous snapshot is non-primary"
+                   << dendl;
+        return false;
+      }
+      if (demoted) {
+        lderr(cct) << "trying to create primary demoted snapshot "
+                   << "when previous snapshot is non-primary"
+                   << dendl;
+        return false;
+      }
+
+      if (requires_orphan != nullptr) {
+        *requires_orphan = !mirror_ns->is_demoted();
+      }
+      if (!mirror_ns->complete) {
+        ldout(cct, 20) << "needs rollback" << dendl;
+        if (!rollback_snap_id) {
+          lderr(cct) << "trying to create primary snapshot "
+                     << "when previous non-primary snapshot is not copied yet"
+                     << dendl;
+          return false;
+        }
+        if (!get_rollback_snap_id(++it, image_ctx->snap_info.rend(),
+                                  rollback_snap_id)) {
+          lderr(cct) << "cannot rollback" << dendl;
+          return false;
+        }
+        ldout(cct, 20) << "rollback_snap_id=" << *rollback_snap_id << dendl;
+      }
+      return true;
+    }
+
+    return true;
+  }
+
+  ldout(cct, 20) << "no previous mirror snapshots found" << dendl;
+  return true;
+}
+
+template <typename I>
+bool can_create_non_primary_snapshot(I *image_ctx) {
+  CephContext *cct = image_ctx->cct;
+
+  std::shared_lock image_locker{image_ctx->image_lock};
+
+  for (auto it = image_ctx->snap_info.rbegin();
+       it != image_ctx->snap_info.rend(); it++) {
+    auto mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (mirror_ns != nullptr) {
+      ldout(cct, 20) << "previous mirror snapshot snap_id=" << it->first << " "
+                     << *mirror_ns << dendl;
+
+      if (mirror_ns->state == cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY) {
+        if (!mirror_ns->complete) {
+          lderr(cct) << "trying to create non-primary snapshot "
+                     << "when previous non-primary snapshot is not copied yet"
+                     << dendl;
+          return false;
+        }
+        return true;
+      }
+
+      if (mirror_ns->state == cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY) {
+        lderr(cct) << "trying to create non-primary snapshot "
+                   << "when previous primary snapshot is not in demoted state"
+                   << dendl;
+        return false;
+      }
+      return true;
+    }
+  }
+
+  ldout(cct, 20) << "no previous mirror snapshots found" << dendl;
+  return true;
+}
+
+template <typename I>
+std::string image_state_object_name(I *image_ctx, uint64_t snap_id,
+                                    uint64_t index) {
+  return IMAGE_STATE_OBJECT_PREFIX + image_ctx->id + "." +
+    stringify(snap_id) + "." + stringify(index);
+}
+
+} // namespace util
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template bool librbd::mirror::snapshot::util::can_create_primary_snapshot(
+  librbd::ImageCtx *image_ctx, bool demoted, bool force,
+  bool* requires_orphan, uint64_t *rollback_snap_id);
+
+template bool librbd::mirror::snapshot::util::can_create_non_primary_snapshot(
+  librbd::ImageCtx *image_ctx);
+
+template std::string librbd::mirror::snapshot::util::image_state_object_name(
+  librbd::ImageCtx *image_ctx, uint64_t snap_id, uint64_t index);
diff --git a/src/librbd/mirror/snapshot/Utils.h b/src/librbd/mirror/snapshot/Utils.h
new file mode 100644
index 000000000..127ec5865
--- /dev/null
+++ b/src/librbd/mirror/snapshot/Utils.h
@@ -0,0 +1,38 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_UTILS_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_UTILS_H
+
+#include "include/int_types.h"
+#include "include/stringify.h"
+#include <string>
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+namespace util {
+
+std::string get_image_meta_key(const std::string& mirror_uuid);
+
+template <typename ImageCtxT = librbd::ImageCtx>
+bool can_create_primary_snapshot(ImageCtxT *image_ctx, bool demoted, bool force,
+                                 bool* requires_orphan,
+                                 uint64_t *rollback_snap_id);
+
+template <typename ImageCtxT = librbd::ImageCtx>
+bool can_create_non_primary_snapshot(ImageCtxT *image_ctx);
+
+template <typename ImageCtxT = librbd::ImageCtx>
+std::string image_state_object_name(ImageCtxT *image_ctx, uint64_t snap_id,
+                                    uint64_t index);
+
+} // namespace util
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_UTILS_H
diff --git a/src/librbd/mirror/snapshot/WriteImageStateRequest.cc b/src/librbd/mirror/snapshot/WriteImageStateRequest.cc
new file mode 100644
index 000000000..c79dd7e2c
--- /dev/null
+++ b/src/librbd/mirror/snapshot/WriteImageStateRequest.cc
@@ -0,0 +1,120 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/mirror/snapshot/WriteImageStateRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/mirror/snapshot/Utils.h"
+
+#define dout_subsys ceph_subsys_rbd
+
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::mirror::snapshot::WriteImageStateRequest: " \
+                           << this << " " << __func__ << ": "
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+namespace {
+
+static size_t header_length() {
+  bufferlist bl;
+  ImageStateHeader header;
+
+  using ceph::encode;
+  encode(header, bl);
+
+  return bl.length();
+}
+
+}
+using librbd::util::create_rados_callback;
+
+template <typename I>
+WriteImageStateRequest<I>::WriteImageStateRequest(I *image_ctx,
+                                                  uint64_t snap_id,
+                                                  const ImageState &image_state,
+                                                  Context *on_finish)
+  : m_image_ctx(image_ctx), m_snap_id(snap_id), m_image_state(image_state),
+    m_on_finish(on_finish), m_object_size(
+      1 << image_ctx->config.template get_val<uint64_t>("rbd_default_order")) {
+  bufferlist bl;
+  encode(m_image_state, bl);
+
+  m_object_count = 1 + (header_length() + bl.length()) / m_object_size;
+  ImageStateHeader header(m_object_count);
+
+  encode(header, m_bl);
+  m_bl.claim_append(bl);
+}
+
+template <typename I>
+void WriteImageStateRequest<I>::send() {
+  write_object();
+}
+
+template <typename I>
+void WriteImageStateRequest<I>::write_object() {
+  CephContext *cct = m_image_ctx->cct;
+  ceph_assert(m_object_count > 0);
+
+  m_object_count--;
+
+  auto oid = util::image_state_object_name(m_image_ctx, m_snap_id,
+                                           m_object_count);
+  ldout(cct, 15) << oid << dendl;
+
+  size_t off = m_object_count * m_object_size;
+  size_t len = std::min(m_bl.length() - off, m_object_size);
+  bufferlist bl;
+  bl.substr_of(m_bl, off, len);
+
+  librados::ObjectWriteOperation op;
+  op.write_full(bl);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    WriteImageStateRequest<I>,
+    &WriteImageStateRequest<I>::handle_write_object>(this);
+  int r = m_image_ctx->md_ctx.aio_operate(oid, comp, &op);
+  ceph_assert(r == 0);
+  comp->release();
+}
+
+template <typename I>
+void WriteImageStateRequest<I>::handle_write_object(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to write object: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_object_count == 0) {
+    finish(0);
+    return;
+  }
+
+  write_object();
+}
+
+template <typename I>
+void WriteImageStateRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 15) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+template class librbd::mirror::snapshot::WriteImageStateRequest<librbd::ImageCtx>;
diff --git a/src/librbd/mirror/snapshot/WriteImageStateRequest.h b/src/librbd/mirror/snapshot/WriteImageStateRequest.h
new file mode 100644
index 000000000..d2c4a7f80
--- /dev/null
+++ b/src/librbd/mirror/snapshot/WriteImageStateRequest.h
@@ -0,0 +1,73 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRROR_SNAPSHOT_WRITE_IMAGE_STATE_REQUEST_H
+#define CEPH_LIBRBD_MIRROR_SNAPSHOT_WRITE_IMAGE_STATE_REQUEST_H
+
+#include "librbd/mirror/snapshot/Types.h"
+
+#include <map>
+#include <string>
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace mirror {
+namespace snapshot {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class WriteImageStateRequest {
+public:
+  static WriteImageStateRequest *create(ImageCtxT *image_ctx, uint64_t snap_id,
+                                        const ImageState &image_state,
+                                        Context *on_finish) {
+    return new WriteImageStateRequest(image_ctx, snap_id, image_state,
+                                      on_finish);
+  }
+
+  WriteImageStateRequest(ImageCtxT *image_ctx, uint64_t snap_id,
+                         const ImageState &image_state, Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * WRITE_OBJECT (repeat for
+   *    |          every object)
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  uint64_t m_snap_id;
+  ImageState m_image_state;
+  Context *m_on_finish;
+
+  bufferlist m_bl;
+
+  const size_t m_object_size;
+  size_t m_object_count = 0;
+
+  void write_object();
+  void handle_write_object(int r);
+
+  void finish(int r);
+};
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+extern template class librbd::mirror::snapshot::WriteImageStateRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_MIRROR_SNAPSHOT_WRITE_IMAGE_STATE_REQUEST_H
diff --git a/src/librbd/mirroring_watcher/Types.cc b/src/librbd/mirroring_watcher/Types.cc
new file mode 100644
index 000000000..3226b6352
--- /dev/null
+++ b/src/librbd/mirroring_watcher/Types.cc
@@ -0,0 +1,136 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "common/Formatter.h"
+#include "include/ceph_assert.h"
+#include "include/stringify.h"
+#include "librbd/mirroring_watcher/Types.h"
+#include "librbd/watcher/Utils.h"
+
+namespace librbd {
+namespace mirroring_watcher {
+
+namespace {
+
+class DumpPayloadVisitor : public boost::static_visitor<void> {
+public:
+  explicit DumpPayloadVisitor(Formatter *formatter) : m_formatter(formatter) {}
+
+  template <typename Payload>
+  inline void operator()(const Payload &payload) const {
+    NotifyOp notify_op = Payload::NOTIFY_OP;
+    m_formatter->dump_string("notify_op", stringify(notify_op));
+    payload.dump(m_formatter);
+  }
+
+private:
+  ceph::Formatter *m_formatter;
+};
+
+} // anonymous namespace
+
+void ModeUpdatedPayload::encode(bufferlist &bl) const {
+  using ceph::encode;
+  encode(static_cast<uint32_t>(mirror_mode), bl);
+}
+
+void ModeUpdatedPayload::decode(__u8 version, bufferlist::const_iterator &iter) {
+  using ceph::decode;
+  uint32_t mirror_mode_decode;
+  decode(mirror_mode_decode, iter);
+  mirror_mode = static_cast<cls::rbd::MirrorMode>(mirror_mode_decode);
+}
+
+void ModeUpdatedPayload::dump(Formatter *f) const {
+  f->dump_stream("mirror_mode") << mirror_mode;
+}
+
+void ImageUpdatedPayload::encode(bufferlist &bl) const {
+  using ceph::encode;
+  encode(static_cast<uint32_t>(mirror_image_state), bl);
+  encode(image_id, bl);
+  encode(global_image_id, bl);
+}
+
+void ImageUpdatedPayload::decode(__u8 version, bufferlist::const_iterator &iter) {
+  using ceph::decode;
+  uint32_t mirror_image_state_decode;
+  decode(mirror_image_state_decode, iter);
+  mirror_image_state = static_cast<cls::rbd::MirrorImageState>(
+    mirror_image_state_decode);
+  decode(image_id, iter);
+  decode(global_image_id, iter);
+}
+
+void ImageUpdatedPayload::dump(Formatter *f) const {
+  f->dump_stream("mirror_image_state") << mirror_image_state;
+  f->dump_string("image_id", image_id);
+  f->dump_string("global_image_id", global_image_id);
+}
+
+void UnknownPayload::encode(bufferlist &bl) const {
+  ceph_abort();
+}
+
+void UnknownPayload::decode(__u8 version, bufferlist::const_iterator &iter) {
+}
+
+void UnknownPayload::dump(Formatter *f) const {
+}
+
+void NotifyMessage::encode(bufferlist& bl) const {
+  ENCODE_START(1, 1, bl);
+  boost::apply_visitor(watcher::util::EncodePayloadVisitor(bl), payload);
+  ENCODE_FINISH(bl);
+}
+
+void NotifyMessage::decode(bufferlist::const_iterator& iter) {
+  DECODE_START(1, iter);
+
+  uint32_t notify_op;
+  decode(notify_op, iter);
+
+  // select the correct payload variant based upon the encoded op
+  switch (notify_op) {
+  case NOTIFY_OP_MODE_UPDATED:
+    payload = ModeUpdatedPayload();
+    break;
+  case NOTIFY_OP_IMAGE_UPDATED:
+    payload = ImageUpdatedPayload();
+    break;
+  default:
+    payload = UnknownPayload();
+    break;
+  }
+
+  apply_visitor(watcher::util::DecodePayloadVisitor(struct_v, iter), payload);
+  DECODE_FINISH(iter);
+}
+
+void NotifyMessage::dump(Formatter *f) const {
+  apply_visitor(DumpPayloadVisitor(f), payload);
+}
+
+void NotifyMessage::generate_test_instances(std::list<NotifyMessage *> &o) {
+  o.push_back(new NotifyMessage(ModeUpdatedPayload(cls::rbd::MIRROR_MODE_DISABLED)));
+  o.push_back(new NotifyMessage(ImageUpdatedPayload(cls::rbd::MIRROR_IMAGE_STATE_DISABLING,
+                                                    "image id", "global image id")));
+}
+
+std::ostream &operator<<(std::ostream &out, const NotifyOp &op) {
+  switch (op) {
+  case NOTIFY_OP_MODE_UPDATED:
+    out << "ModeUpdated";
+    break;
+  case NOTIFY_OP_IMAGE_UPDATED:
+    out << "ImageUpdated";
+    break;
+  default:
+    out << "Unknown (" << static_cast<uint32_t>(op) << ")";
+    break;
+  }
+  return out;
+}
+
+} // namespace mirroring_watcher
+} // namespace librbd
diff --git a/src/librbd/mirroring_watcher/Types.h b/src/librbd/mirroring_watcher/Types.h
new file mode 100644
index 000000000..1e096a9d3
--- /dev/null
+++ b/src/librbd/mirroring_watcher/Types.h
@@ -0,0 +1,102 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_MIRRORING_WATCHER_TYPES_H
+#define CEPH_LIBRBD_MIRRORING_WATCHER_TYPES_H
+
+#include "include/int_types.h"
+#include "include/buffer_fwd.h"
+#include "include/encoding.h"
+#include "cls/rbd/cls_rbd_types.h"
+#include <iosfwd>
+#include <list>
+#include <string>
+#include <boost/variant.hpp>
+
+namespace ceph { class Formatter; }
+
+namespace librbd {
+namespace mirroring_watcher {
+
+enum NotifyOp {
+  NOTIFY_OP_MODE_UPDATED  = 0,
+  NOTIFY_OP_IMAGE_UPDATED = 1
+};
+
+struct ModeUpdatedPayload {
+  static const NotifyOp NOTIFY_OP = NOTIFY_OP_MODE_UPDATED;
+
+  cls::rbd::MirrorMode mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
+
+  ModeUpdatedPayload() {
+  }
+  ModeUpdatedPayload(cls::rbd::MirrorMode mirror_mode)
+    : mirror_mode(mirror_mode) {
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(__u8 version, bufferlist::const_iterator &iter);
+  void dump(Formatter *f) const;
+};
+
+struct ImageUpdatedPayload {
+  static const NotifyOp NOTIFY_OP = NOTIFY_OP_IMAGE_UPDATED;
+
+  cls::rbd::MirrorImageState mirror_image_state =
+    cls::rbd::MIRROR_IMAGE_STATE_ENABLED;
+  std::string image_id;
+  std::string global_image_id;
+
+  ImageUpdatedPayload() {
+  }
+  ImageUpdatedPayload(cls::rbd::MirrorImageState mirror_image_state,
+                      const std::string &image_id,
+                      const std::string &global_image_id)
+    : mirror_image_state(mirror_image_state), image_id(image_id),
+      global_image_id(global_image_id) {
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(__u8 version, bufferlist::const_iterator &iter);
+  void dump(Formatter *f) const;
+};
+
+struct UnknownPayload {
+  static const NotifyOp NOTIFY_OP = static_cast<NotifyOp>(-1);
+
+  UnknownPayload() {
+  }
+
+  void encode(bufferlist &bl) const;
+  void decode(__u8 version, bufferlist::const_iterator &iter);
+  void dump(Formatter *f) const;
+};
+
+typedef boost::variant<ModeUpdatedPayload,
+                       ImageUpdatedPayload,
+                       UnknownPayload> Payload;
+
+struct NotifyMessage {
+  NotifyMessage(const Payload &payload = UnknownPayload()) : payload(payload) {
+  }
+
+  Payload payload;
+
+  void encode(bufferlist& bl) const;
+  void decode(bufferlist::const_iterator& it);
+  void dump(Formatter *f) const;
+
+  static void generate_test_instances(std::list<NotifyMessage *> &o);
+};
+
+WRITE_CLASS_ENCODER(NotifyMessage);
+
+std::ostream &operator<<(std::ostream &out, const NotifyOp &op);
+
+} // namespace mirroring_watcher
+} // namespace librbd
+
+using librbd::mirroring_watcher::encode;
+using librbd::mirroring_watcher::decode;
+
+#endif // CEPH_LIBRBD_MIRRORING_WATCHER_TYPES_H