librbd: make CreatePrimaryRequest remove any unlinked mirror snapshots

After commit ac552c9 ("librbd: localize snap_remove op for mirror
snapshots"), rbd-mirror daemon no longer removes mirror snapshots when
it's done syncing them -- instead it only unlinks from them.  However,
CreatePrimaryRequest state machine was not adjusted to compensate and
hence two cases were missed:

- primary demotion snapshot (rbd-mirror daemon unlinks from primary
  demotion snapshots just like it does from regular primary snapshots);
  this comes up when an image is demoted but then promoted on the same
  cluster

- non-primary demotion snapshot (unlike regular non-primary snapshots,
  non-primary demotion snapshots store peer uuids and rbd-mirror daemon
  does unlinking just like in the case of primary snapshots); this
  comes up when an image is demoted and promoted on the other cluster

Related is the case of orphan snapshots.  Since they are dummy to begin
with, CreatePrimaryRequest would now clean up the orphan snapshot after
the creation of the force promote snapshot.

Fixes: https://tracker.ceph.com/issues/61707
Co-authored-by: Christopher Hoffman <[email protected]>
Signed-off-by: Ilya Dryomov <[email protected]>

Author: idryomov

src/librbd/mirror/snapshot/CreatePrimaryRequest.cc

@@ -177,62 +177,69 @@ void CreatePrimaryRequest<I>::handle_refresh_image(int r) {
 
 template <typename I>
 void CreatePrimaryRequest<I>::unlink_peer() {
+  // TODO: Document semantics for 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 = std::get_if<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.empty() &&
-          info->mirror_peer_uuids.count(peer) == 0) {
-        continue;
-      }
-      if (info->mirror_peer_uuids.empty() || !info->complete) {
-        peer_uuid = peer;
-        snap_id = snap_it.first;
-        break;
-      }
-      count++;
-      if (count == max_snapshots) {
-        unlink_snap_id = snap_it.first;
-      }
-      if (count > max_snapshots) {
-        peer_uuid = peer;
-        snap_id = unlink_snap_id;
-        break;
+    for (const auto& peer : m_mirror_peer_uuids) {
+      for (const auto& snap_info_pair : m_image_ctx->snap_info) {
+        auto info = std::get_if<cls::rbd::MirrorSnapshotNamespace>(
+          &snap_info_pair.second.snap_namespace);
+        if (info == nullptr) {
+          continue;
+        }
+        if (info->mirror_peer_uuids.empty() ||
+            (info->mirror_peer_uuids.count(peer) != 0 &&
+             info->is_primary() && !info->complete)) {
+          peer_uuid = peer;
+          snap_id = snap_info_pair.first;
+          goto do_unlink;
+        }
       }
     }
-    if (snap_id != CEPH_NOSNAP) {
-      break;
+    for (const auto& peer : m_mirror_peer_uuids) {
+      size_t count = 0;
+      uint64_t unlink_snap_id = 0;
+      for (const auto& snap_info_pair : m_image_ctx->snap_info) {
+        auto info = std::get_if<cls::rbd::MirrorSnapshotNamespace>(
+          &snap_info_pair.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;
+        }
+        if (info->mirror_peer_uuids.count(peer) == 0) {
+          // snapshot is not linked with this peer
+          continue;
+        }
+        count++;
+        if (count == max_snapshots) {
+          unlink_snap_id = snap_info_pair.first;
+        }
+        if (count > max_snapshots) {
+          peer_uuid = peer;
+          snap_id = unlink_snap_id;
+          goto do_unlink;
+        }
+      }
     }
   }
 
-  if (snap_id == CEPH_NOSNAP) {
-    finish(0);
-    return;
-  }
+  finish(0);
+  return;
 
+do_unlink:
   CephContext *cct = m_image_ctx->cct;
   ldout(cct, 15) << "peer=" << peer_uuid << ", snap_id=" << snap_id << dendl;