From 2bfac34784645b4ba895c7613cccd124a085d9a7 Mon Sep 17 00:00:00 2001
From: Mykola Golub <mgolub@suse.com>
Date: Thu, 10 Oct 2019 16:18:00 +0100
Subject: [PATCH] librbd: state machine for creating mirroring snapshot

Signed-off-by: Mykola Golub <mgolub@suse.com>
---
 src/common/options.cc                         |   5 +
 src/librbd/CMakeLists.txt                     |   2 +
 .../snapshot/CreateNonPrimaryRequest.cc       | 203 +++++++++
 .../mirror/snapshot/CreateNonPrimaryRequest.h |  99 +++++
 .../mirror/snapshot/CreatePrimaryRequest.cc   | 343 +++++++++++++++
 .../mirror/snapshot/CreatePrimaryRequest.h    | 104 +++++
 src/test/librbd/CMakeLists.txt                |   2 +
 .../test_mock_CreateNonPrimaryRequest.cc      | 272 ++++++++++++
 .../test_mock_CreatePrimaryRequest.cc         | 405 ++++++++++++++++++
 9 files changed, 1435 insertions(+)
 create mode 100644 src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc
 create mode 100644 src/librbd/mirror/snapshot/CreateNonPrimaryRequest.h
 create mode 100644 src/librbd/mirror/snapshot/CreatePrimaryRequest.cc
 create mode 100644 src/librbd/mirror/snapshot/CreatePrimaryRequest.h
 create mode 100644 src/test/librbd/mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc
 create mode 100644 src/test/librbd/mirror/snapshot/test_mock_CreatePrimaryRequest.cc

diff --git a/src/common/options.cc b/src/common/options.cc
index 384df54dc5f..2b57613faa8 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -7175,6 +7175,11 @@ static std::vector<Option> get_rbd_options() {
     .set_default(0)
     .set_description("time-delay in seconds for rbd-mirror asynchronous replication"),
 
+    Option("rbd_mirroring_max_mirroring_snapshots", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+    .set_default(3)
+    .set_min(3)
+    .set_description("mirroring snapshots limit"),
+
     Option("rbd_default_format", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
     .set_default(2)
     .set_description("default image format for new images"),
diff --git a/src/librbd/CMakeLists.txt b/src/librbd/CMakeLists.txt
index ee869e66014..9bf456f8781 100644
--- a/src/librbd/CMakeLists.txt
+++ b/src/librbd/CMakeLists.txt
@@ -104,6 +104,8 @@ set(librbd_internal_srcs
   mirror/GetInfoRequest.cc
   mirror/GetStatusRequest.cc
   mirror/PromoteRequest.cc
+  mirror/snapshot/CreateNonPrimaryRequest.cc
+  mirror/snapshot/CreatePrimaryRequest.cc
   mirror/snapshot/UnlinkPeerRequest.cc
   object_map/CreateRequest.cc
   object_map/InvalidateRequest.cc
diff --git a/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc
new file mode 100644
index 00000000000..305cb32d666
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.cc
@@ -0,0 +1,203 @@
+// -*- 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"
+
+#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>
+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, 20) << 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, 20) << "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, 20) << 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, 20) << "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 (!validate_snapshot()) {
+    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();
+
+  create_snapshot();
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::create_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  cls::rbd::MirrorNonPrimarySnapshotNamespace ns{m_primary_mirror_uuid,
+                                                 m_primary_snap_id};
+  auto ctx = create_context_callback<
+    CreateNonPrimaryRequest<I>,
+    &CreateNonPrimaryRequest<I>::handle_create_snapshot>(this);
+  m_image_ctx->operations->snap_create(ns, m_snap_name, ctx);
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::handle_create_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create mirror snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_snap_id != nullptr) {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+    cls::rbd::MirrorNonPrimarySnapshotNamespace ns{m_primary_mirror_uuid,
+                                                   m_primary_snap_id};
+    *m_snap_id = m_image_ctx->get_snap_id(ns, m_snap_name);
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void CreateNonPrimaryRequest<I>::finish(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+template <typename I>
+bool CreateNonPrimaryRequest<I>::validate_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+
+  std::shared_lock image_locker{m_image_ctx->image_lock};
+
+  for (auto it = m_image_ctx->snap_info.rbegin();
+       it != m_image_ctx->snap_info.rend(); it++) {
+    auto primary = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (primary != nullptr) {
+      ldout(cct, 20) << "previous mirror snapshot snap_id=" << it->first << " "
+                     << *primary << dendl;
+      if (!primary->demoted) {
+        lderr(cct) << "trying to create non-primary snapshot "
+                   << "when previous primary snapshot is not in demoted state"
+                   << dendl;
+        return false;
+      }
+      return true;
+    }
+    auto non_primary = boost::get<cls::rbd::MirrorNonPrimarySnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (non_primary == nullptr) {
+      continue;
+    }
+    ldout(cct, 20) << "previous snapshot snap_id=" << it->first << " "
+                   << *non_primary << dendl;
+    if (!non_primary->copied) {
+      lderr(cct) << "trying to create non-primary snapshot "
+                 << "when previous non-primary snapshot is not copied yet"
+                 << dendl;
+      return false;
+    }
+    return true;
+  }
+
+  ldout(cct, 20) << "no previous mirror snapshots found" << dendl;
+  return true;
+}
+
+} // 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 00000000000..6d916c47414
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreateNonPrimaryRequest.h
@@ -0,0 +1,99 @@
+// -*- 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 <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,
+                                         const std::string &primary_mirror_uuid,
+                                         uint64_t primary_snap_id,
+                                         uint64_t *snap_id,
+                                         Context *on_finish) {
+    return new CreateNonPrimaryRequest(image_ctx, primary_mirror_uuid,
+                                       primary_snap_id, snap_id, on_finish);
+  }
+
+  CreateNonPrimaryRequest(ImageCtxT *image_ctx,
+                          const std::string &primary_mirror_uuid,
+                          uint64_t primary_snap_id, uint64_t *snap_id,
+                          Context *on_finish)
+    : m_image_ctx(image_ctx), m_primary_mirror_uuid(primary_mirror_uuid),
+      m_primary_snap_id(primary_snap_id), m_snap_id(snap_id),
+      m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * REFRESH_IMAGE
+   *    |
+   *    v
+   * GET_MIRROR_IMAGE
+   *    |
+   *    v
+   * CREATE_SNAPSHOT
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  std::string m_primary_mirror_uuid;
+  uint64_t m_primary_snap_id;
+  uint64_t *m_snap_id;
+  Context *m_on_finish;
+
+  std::string m_snap_name;
+
+  bufferlist m_out_bl;
+
+  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 finish(int r);
+
+  bool validate_snapshot();
+};
+
+} // 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 00000000000..d325667373b
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreatePrimaryRequest.cc
@@ -0,0 +1,343 @@
+// -*- 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"
+
+#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>
+void CreatePrimaryRequest<I>::send() {
+  refresh_image();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::refresh_image() {
+  if (!m_image_ctx->state->is_refresh_required()) {
+    get_mirror_image();
+    return;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << 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, 20) << "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 CreatePrimaryRequest<I>::get_mirror_image() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  librados::ObjectReadOperation op;
+  cls_client::mirror_image_get_start(&op, m_image_ctx->id);
+
+  librados::AioCompletion *comp = create_rados_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<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 CreatePrimaryRequest<I>::handle_get_mirror_image(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "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 ||
+      mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
+    lderr(cct) << "snapshot based mirroring is not enabled" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  if (!validate_snapshot()) {
+    finish(-EINVAL);
+    return;
+  }
+
+  uuid_d uuid_gen;
+  uuid_gen.generate_random();
+  m_snap_name = ".mirror.primary." + mirror_image.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, 20) << 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_image_ctx->md_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, 20) << "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()) {
+    lderr(cct) << "no mirror tx peers configured for the pool" << dendl;
+    finish(-EINVAL);
+    return;
+  }
+
+  create_snapshot();
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::create_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  cls::rbd::MirrorPrimarySnapshotNamespace ns{m_demoted, m_mirror_peer_uuids};
+  auto ctx = create_context_callback<
+    CreatePrimaryRequest<I>,
+    &CreatePrimaryRequest<I>::handle_create_snapshot>(this);
+  m_image_ctx->operations->snap_create(ns, m_snap_name, ctx);
+}
+
+template <typename I>
+void CreatePrimaryRequest<I>::handle_create_snapshot(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to create mirror snapshot: " << cpp_strerror(r)
+               << dendl;
+    finish(r);
+    return;
+  }
+
+  if (m_snap_id != nullptr) {
+    std::shared_lock image_locker{m_image_ctx->image_lock};
+    cls::rbd::MirrorPrimarySnapshotNamespace ns{m_demoted, m_mirror_peer_uuids};
+    *m_snap_id = m_image_ctx->get_snap_id(ns, m_snap_name);
+  }
+
+  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 prev_snap_id = 0;
+    for (auto &snap_it : m_image_ctx->snap_info) {
+      if (boost::get<cls::rbd::MirrorNonPrimarySnapshotNamespace>(
+            &snap_it.second.snap_namespace)) {
+        // reset counters -- we count primary snapshots after the last promotion
+        count = 0;
+        prev_snap_id = 0;
+        continue;
+      }
+      auto info = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+        &snap_it.second.snap_namespace);
+      if (info == nullptr) {
+        continue;
+      }
+      if (info->demoted) {
+        // reset counters -- we count primary snapshots after the last promotion
+        count = 0;
+        prev_snap_id = 0;
+        continue;
+      }
+      if (info->mirror_peer_uuids.count(peer) == 0) {
+        continue;
+      }
+      count++;
+      if (count == max_snapshots - 1) {
+        prev_snap_id = snap_it.first;
+      } else if (count > max_snapshots) {
+        peer_uuid = peer;
+        snap_id = prev_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, 20) << "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, 20) << "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, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+template <typename I>
+bool CreatePrimaryRequest<I>::validate_snapshot() {
+  CephContext *cct = m_image_ctx->cct;
+
+  std::shared_lock image_locker{m_image_ctx->image_lock};
+
+  for (auto it = m_image_ctx->snap_info.rbegin();
+       it != m_image_ctx->snap_info.rend(); it++) {
+    auto non_primary = boost::get<cls::rbd::MirrorNonPrimarySnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (non_primary != nullptr) {
+      ldout(cct, 20) << "previous mirror snapshot snap_id=" << it->first << " "
+                     << *non_primary << dendl;
+      if (!m_force) {
+        lderr(cct) << "trying to create primary snapshot without force "
+                   << "when previous snapshot is non-primary"
+                   << dendl;
+        return false;
+      }
+      if (m_demoted) {
+        lderr(cct) << "trying to create primary demoted snapshot "
+                   << "when previous snapshot is non-primary"
+                   << dendl;
+        return false;
+      }
+      if (!non_primary->copied) {
+        lderr(cct) << "trying to create primary snapshot "
+                   << "when previous non-primary snapshot is not copied yet"
+                   << dendl;
+        return false;
+      }
+      return true;
+    }
+    auto primary = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+      &it->second.snap_namespace);
+    if (primary == nullptr) {
+      continue;
+    }
+    ldout(cct, 20) << "previous snapshot snap_id=" << it->first << " "
+                   << *primary << dendl;
+    if (primary->demoted && !m_force) {
+      lderr(cct) << "trying to create primary snapshot without force "
+                 << "when previous primary snapshot is demoted"
+                 << dendl;
+      return false;
+    }
+    return true;
+  }
+
+  ldout(cct, 20) << "no previous mirror snapshots found" << dendl;
+  return true;
+}
+
+} // 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 00000000000..9f0b951b723
--- /dev/null
+++ b/src/librbd/mirror/snapshot/CreatePrimaryRequest.h
@@ -0,0 +1,104 @@
+// -*- 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 "cls/rbd/cls_rbd_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, bool demoted,
+                                      bool force, uint64_t *snap_id,
+                                      Context *on_finish) {
+    return new CreatePrimaryRequest(image_ctx, demoted, force, snap_id,
+                                    on_finish);
+  }
+
+  CreatePrimaryRequest(ImageCtxT *image_ctx, bool demoted, bool force,
+                       uint64_t *snap_id, Context *on_finish)
+    : m_image_ctx(image_ctx), m_demoted(demoted), m_force(force),
+      m_snap_id(snap_id), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * REFRESH_IMAGE
+   *    |
+   *    v
+   * GET_MIRROR_IMAGE
+   *    |
+   *    v
+   * GET_MIRROR_PEERS
+   *    |
+   *    v
+   * CREATE_SNAPSHOT
+   *    |
+   *    v
+   * UNLINK_PEER (skip if not needed,
+   *    |         repeat if needed)
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  ImageCtxT *m_image_ctx;
+  const bool m_demoted;
+  const bool m_force;
+  uint64_t *m_snap_id;
+  Context *m_on_finish;
+
+  std::set<std::string> m_mirror_peer_uuids;
+  std::string m_snap_name;
+
+  bufferlist m_out_bl;
+
+  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 unlink_peer();
+  void handle_unlink_peer(int r);
+
+  void finish(int r);
+
+  bool validate_snapshot();
+};
+
+} // 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/test/librbd/CMakeLists.txt b/src/test/librbd/CMakeLists.txt
index 7f08342f97d..501881a8a8f 100644
--- a/src/test/librbd/CMakeLists.txt
+++ b/src/test/librbd/CMakeLists.txt
@@ -85,6 +85,8 @@ set(unittest_librbd_srcs
   managed_lock/test_mock_GetLockerRequest.cc
   managed_lock/test_mock_ReacquireRequest.cc
   managed_lock/test_mock_ReleaseRequest.cc
+  mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc
+  mirror/snapshot/test_mock_CreatePrimaryRequest.cc
   mirror/snapshot/test_mock_UnlinkPeerRequest.cc
   mirror/test_mock_DisableRequest.cc
   object_map/test_mock_InvalidateRequest.cc
diff --git a/src/test/librbd/mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc b/src/test/librbd/mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc
new file mode 100644
index 00000000000..9a3a1ab6faa
--- /dev/null
+++ b/src/test/librbd/mirror/snapshot/test_mock_CreateNonPrimaryRequest.cc
@@ -0,0 +1,272 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockOperations.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "librbd/mirror/snapshot/CreateNonPrimaryRequest.h"
+
+namespace librbd {
+
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  explicit MockTestImageCtx(librbd::ImageCtx& image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+} // namespace librbd
+
+// template definitions
+#include "librbd/mirror/snapshot/CreateNonPrimaryRequest.cc"
+template class librbd::mirror::snapshot::CreateNonPrimaryRequest<librbd::MockTestImageCtx>;
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::InSequence;
+using ::testing::Return;
+using ::testing::StrEq;
+using ::testing::WithArg;
+
+class TestMockMirrorSnapshotCreateNonPrimaryRequest : public TestMockFixture {
+public:
+  typedef CreateNonPrimaryRequest<MockTestImageCtx> MockCreateNonPrimaryRequest;
+
+  uint64_t m_snap_seq = 0;
+
+  void snap_create(MockTestImageCtx &mock_image_ctx,
+                   const cls::rbd::SnapshotNamespace &ns,
+                   const std::string& snap_name) {
+    ASSERT_TRUE(mock_image_ctx.snap_info.insert(
+                  {m_snap_seq++,
+                   SnapInfo{snap_name, ns, 0, {}, 0, 0, {}}}).second);
+  }
+
+  void expect_refresh_image(MockTestImageCtx &mock_image_ctx,
+                            bool refresh_required, int r) {
+    EXPECT_CALL(*mock_image_ctx.state, is_refresh_required())
+      .WillOnce(Return(refresh_required));
+    if (refresh_required) {
+      EXPECT_CALL(*mock_image_ctx.state, refresh(_))
+        .WillOnce(CompleteContext(r, mock_image_ctx.image_ctx->op_work_queue));
+    }
+  }
+
+  void expect_get_mirror_image(MockTestImageCtx &mock_image_ctx,
+                               const cls::rbd::MirrorImage &mirror_image,
+                               int r) {
+    using ceph::encode;
+    bufferlist bl;
+    encode(mirror_image, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(mock_image_ctx.md_ctx),
+                exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_image_get"),
+                     _, _, _))
+      .WillOnce(DoAll(WithArg<5>(CopyInBufferlist(bl)),
+                      Return(r)));
+  }
+
+  void expect_create_snapshot(MockTestImageCtx &mock_image_ctx, int r) {
+    EXPECT_CALL(*mock_image_ctx.operations, snap_create(_, _, _))
+      .WillOnce(WithArg<2>(CompleteContext(
+                             r, mock_image_ctx.image_ctx->op_work_queue)));
+  }
+};
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, Success) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, RefreshError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, GetMirrorImageError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, CreateSnapshotError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_create_snapshot(mock_image_ctx, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, ValidateErrorPrimary) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorPrimarySnapshotNamespace ns{false, {"peer_uuid"}};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, ValidatePrimaryDemoted) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorPrimarySnapshotNamespace ns{true, {"peer_uuid"}};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, ValidateErrorNonPrimaryNotCopied) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorNonPrimarySnapshotNamespace ns{"mirror_uuid", 111};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreateNonPrimaryRequest, ValidateNonPrimaryCopied) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorNonPrimarySnapshotNamespace ns{"mirror_uuid", 111};
+  ns.copied = true;
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreateNonPrimaryRequest(&mock_image_ctx, "mirror_uuid",
+                                             123, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
diff --git a/src/test/librbd/mirror/snapshot/test_mock_CreatePrimaryRequest.cc b/src/test/librbd/mirror/snapshot/test_mock_CreatePrimaryRequest.cc
new file mode 100644
index 00000000000..6b8b1b52f3b
--- /dev/null
+++ b/src/test/librbd/mirror/snapshot/test_mock_CreatePrimaryRequest.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 "include/stringify.h"
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockOperations.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
+#include "librbd/mirror/snapshot/UnlinkPeerRequest.h"
+
+namespace librbd {
+
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  explicit MockTestImageCtx(librbd::ImageCtx& image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace mirror {
+namespace snapshot {
+
+template <>
+struct UnlinkPeerRequest<MockTestImageCtx> {
+  uint64_t snap_id = CEPH_NOSNAP;
+  std::string mirror_peer_uuid;
+  Context* on_finish = nullptr;
+  static UnlinkPeerRequest* s_instance;
+  static UnlinkPeerRequest *create(MockTestImageCtx *image_ctx,
+                                   uint64_t snap_id,
+                                   const std::string &mirror_peer_uuid,
+                                   Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->snap_id = snap_id;
+    s_instance->mirror_peer_uuid = mirror_peer_uuid;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  MOCK_METHOD0(send, void());
+
+  UnlinkPeerRequest() {
+    s_instance = this;
+  }
+};
+
+UnlinkPeerRequest<MockTestImageCtx>* UnlinkPeerRequest<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+// template definitions
+#include "librbd/mirror/snapshot/CreatePrimaryRequest.cc"
+template class librbd::mirror::snapshot::CreatePrimaryRequest<librbd::MockTestImageCtx>;
+
+namespace librbd {
+namespace mirror {
+namespace snapshot {
+
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::StrEq;
+using ::testing::WithArg;
+
+class TestMockMirrorSnapshotCreatePrimaryRequest : public TestMockFixture {
+public:
+  typedef CreatePrimaryRequest<MockTestImageCtx> MockCreatePrimaryRequest;
+  typedef UnlinkPeerRequest<MockTestImageCtx> MockUnlinkPeerRequest;
+
+  uint64_t m_snap_seq = 0;
+
+  void snap_create(MockTestImageCtx &mock_image_ctx,
+                   const cls::rbd::SnapshotNamespace &ns,
+                   const std::string& snap_name) {
+    ASSERT_TRUE(mock_image_ctx.snap_info.insert(
+                  {m_snap_seq++,
+                   SnapInfo{snap_name, ns, 0, {}, 0, 0, {}}}).second);
+  }
+
+  void expect_refresh_image(MockTestImageCtx &mock_image_ctx,
+                            bool refresh_required, int r) {
+    EXPECT_CALL(*mock_image_ctx.state, is_refresh_required())
+      .WillOnce(Return(refresh_required));
+    if (refresh_required) {
+      EXPECT_CALL(*mock_image_ctx.state, refresh(_))
+        .WillOnce(CompleteContext(r, mock_image_ctx.image_ctx->op_work_queue));
+    }
+  }
+
+  void expect_get_mirror_image(MockTestImageCtx &mock_image_ctx,
+                               const cls::rbd::MirrorImage &mirror_image,
+                               int r) {
+    using ceph::encode;
+    bufferlist bl;
+    encode(mirror_image, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(mock_image_ctx.md_ctx),
+                exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_image_get"),
+                     _, _, _))
+      .WillOnce(DoAll(WithArg<5>(CopyInBufferlist(bl)),
+                      Return(r)));
+  }
+
+  void expect_get_mirror_peers(MockTestImageCtx &mock_image_ctx,
+                               const std::vector<cls::rbd::MirrorPeer> &peers,
+                               int r) {
+    using ceph::encode;
+    bufferlist bl;
+    encode(peers, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(mock_image_ctx.md_ctx),
+                exec(RBD_MIRRORING, _, StrEq("rbd"), StrEq("mirror_peer_list"),
+                     _, _, _))
+      .WillOnce(DoAll(WithArg<5>(CopyInBufferlist(bl)),
+                      Return(r)));
+  }
+
+  void expect_create_snapshot(MockTestImageCtx &mock_image_ctx, int r) {
+    EXPECT_CALL(*mock_image_ctx.operations, snap_create(_, _, _))
+      .WillOnce(DoAll(
+                  Invoke([this, &mock_image_ctx, r](
+                             const cls::rbd::SnapshotNamespace &ns,
+                             const std::string& snap_name,
+                             Context *on_finish) {
+                           if (r != 0) {
+                             return;
+                           }
+                           snap_create(mock_image_ctx, ns, snap_name);
+                         }),
+                  WithArg<2>(CompleteContext(
+                               r, mock_image_ctx.image_ctx->op_work_queue))
+                  ));
+  }
+
+  void expect_unlink_peer(MockTestImageCtx &mock_image_ctx,
+                          MockUnlinkPeerRequest &mock_unlink_peer_request,
+                          uint64_t snap_id, const std::string &peer_uuid,
+                          int r) {
+    EXPECT_CALL(mock_unlink_peer_request, send())
+      .WillOnce(Invoke([&mock_image_ctx, &mock_unlink_peer_request, snap_id,
+                        peer_uuid, r]() {
+                         ASSERT_EQ(mock_unlink_peer_request.mirror_peer_uuid,
+                                   peer_uuid);
+                         ASSERT_EQ(mock_unlink_peer_request.snap_id, snap_id);
+                         if (r == 0) {
+                           auto it = mock_image_ctx.snap_info.find(snap_id);
+                           ASSERT_NE(it, mock_image_ctx.snap_info.end());
+                           auto info =
+                             boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+                               &it->second.snap_namespace);
+                           ASSERT_NE(nullptr, info);
+                           ASSERT_NE(0, info->mirror_peer_uuids.erase(
+                                       peer_uuid));
+                           if (info->mirror_peer_uuids.empty()) {
+                             mock_image_ctx.snap_info.erase(it);
+                           }
+                         }
+                         mock_image_ctx.image_ctx->op_work_queue->queue(
+                           mock_unlink_peer_request.on_finish, r);
+                       }));
+  }
+};
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, Success) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_get_mirror_peers(mock_image_ctx,
+                          {{"uuid", cls::rbd::MIRROR_PEER_DIRECTION_TX, "ceph",
+                            "mirror", "fsid"}}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, RefreshError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, GetMirrorImageError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, GetMirrorPeersError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_get_mirror_peers(mock_image_ctx,
+                          {{"uuid", cls::rbd::MIRROR_PEER_DIRECTION_TX, "ceph",
+                            "mirror", "fsid"}}, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, CreateSnapshotError) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_get_mirror_peers(mock_image_ctx,
+                          {{"uuid", cls::rbd::MIRROR_PEER_DIRECTION_TX, "ceph",
+                            "mirror", "fsid"}}, 0);
+  expect_create_snapshot(mock_image_ctx, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, ValidateErrorNonPrimaryNotCopied) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorNonPrimarySnapshotNamespace ns{"mirror_uuid", 123};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, true, nullptr,
+                                          &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, ValidateErrorDemotedNonPrimary) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorNonPrimarySnapshotNamespace ns{"mirror_uuid", 123};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, false, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, true, true, nullptr,
+                                          &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, ValidateDemotedPrimary) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorPrimarySnapshotNamespace ns{false, {"uuid"}};
+  snap_create(mock_image_ctx, ns, "mirror_snap");
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_get_mirror_peers(mock_image_ctx,
+                          {{"uuid", cls::rbd::MIRROR_PEER_DIRECTION_TX, "ceph",
+                            "mirror", "fsid"}}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, true, false, nullptr,
+                                          &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockMirrorSnapshotCreatePrimaryRequest, SuccessUnlinkPeer) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->config.set_val("conf_rbd_mirroring_max_mirroring_snapshots", "3");
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  for (int i = 0; i < 3; i++) {
+    cls::rbd::MirrorPrimarySnapshotNamespace ns{false, {"uuid"}};
+    snap_create(mock_image_ctx, ns, "mirror_snap");
+  }
+
+  InSequence seq;
+
+  expect_refresh_image(mock_image_ctx, true, 0);
+  expect_get_mirror_image(
+    mock_image_ctx, {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT, "gid",
+                     cls::rbd::MIRROR_IMAGE_STATE_ENABLED}, 0);
+  expect_get_mirror_peers(mock_image_ctx,
+                          {{"uuid", cls::rbd::MIRROR_PEER_DIRECTION_TX, "ceph",
+                            "mirror", "fsid"}}, 0);
+  expect_create_snapshot(mock_image_ctx, 0);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  auto it = mock_image_ctx.snap_info.rbegin();
+  auto snap_id = (++it)->first;
+  expect_unlink_peer(mock_image_ctx, mock_unlink_peer_request, snap_id, "uuid",
+                     0);
+  C_SaferCond ctx;
+  auto req = new MockCreatePrimaryRequest(&mock_image_ctx, false, false,
+                                          nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
-- 
2.39.5