Merge pull request #7884 from dillaman/wip-14663

librbd: allocate new journal tag after acquiring exclusive lock

Reviewed-by: Josh Durgin <jdurgin@redhat.com>

src/cls/rbd/cls_rbd.cc

@@ -111,6 +111,8 @@ cls_method_handle_t h_old_snapshots_list;
 cls_method_handle_t h_old_snapshot_add;
 cls_method_handle_t h_old_snapshot_remove;
 cls_method_handle_t h_old_snapshot_rename;
+cls_method_handle_t h_mirror_uuid_get;
+cls_method_handle_t h_mirror_uuid_set;
 cls_method_handle_t h_mirror_mode_get;
 cls_method_handle_t h_mirror_mode_set;
 cls_method_handle_t h_mirror_peer_list;
@@ -204,6 +206,15 @@ static int read_key(cls_method_context_t hctx, const string &key, T *out)
   return 0;
 }
 
+static int remove_key(cls_method_context_t hctx, const string &key) {
+  int r = cls_cxx_map_remove_key(hctx, key);
+  if (r < 0 && r != -ENOENT) {
+      CLS_ERR("failed to remove key: %s", key.c_str());
+      return r;
+  }
+  return 0;
+}
+
 static bool is_valid_id(const string &id) {
   if (!id.size())
     return false;
@@ -2939,6 +2950,7 @@ int old_snapshot_rename(cls_method_context_t hctx, bufferlist *in, bufferlist *o
 
 namespace mirror {
 
+static const std::string UUID("mirror_uuid");
 static const std::string MODE("mirror_mode");
 static const std::string PEER_KEY_PREFIX("mirror_peer_");
 static const std::string IMAGE_KEY_PREFIX("image_");
@@ -2951,6 +2963,20 @@ std::string image_key(const string &image_id) {
   return IMAGE_KEY_PREFIX + image_id;
 }
 
+int uuid_get(cls_method_context_t hctx, std::string *mirror_uuid) {
+  bufferlist mirror_uuid_bl;
+  int r = cls_cxx_map_get_val(hctx, mirror::UUID, &mirror_uuid_bl);
+  if (r < 0) {
+    if (r != -ENOENT) {
+      CLS_ERR("error reading mirror uuid: %s", cpp_strerror(r).c_str());
+    }
+    return r;
+  }
+
+  *mirror_uuid = std::string(mirror_uuid_bl.c_str(), mirror_uuid_bl.length());
+  return 0;
+}
+
 int read_peers(cls_method_context_t hctx,
                std::vector<cls::rbd::MirrorPeer> *peers) {
   std::string last_read = PEER_KEY_PREFIX;
@@ -3115,6 +3141,67 @@ int image_remove(cls_method_context_t hctx, const string &image_id) {
 
 } // namespace mirror
 
+/**
+ * Input:
+ * none
+ *
+ * Output:
+ * @param uuid (std::string)
+ * @returns 0 on success, negative error code on failure
+ */
+int mirror_uuid_get(cls_method_context_t hctx, bufferlist *in,
+                    bufferlist *out) {
+  std::string mirror_uuid;
+  int r = mirror::uuid_get(hctx, &mirror_uuid);
+  if (r < 0) {
+    return r;
+  }
+
+  ::encode(mirror_uuid, *out);
+  return 0;
+}
+
+/**
+ * Input:
+ * @param mirror_uuid (std::string)
+ *
+ * Output:
+ * @returns 0 on success, negative error code on failure
+ */
+int mirror_uuid_set(cls_method_context_t hctx, bufferlist *in,
+                    bufferlist *out) {
+  std::string mirror_uuid;
+  try {
+    bufferlist::iterator bl_it = in->begin();
+    ::decode(mirror_uuid, bl_it);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  if (mirror_uuid.empty()) {
+    CLS_ERR("cannot set empty mirror uuid");
+    return -EINVAL;
+  }
+
+  uint32_t mirror_mode;
+  int r = read_key(hctx, mirror::MODE, &mirror_mode);
+  if (r < 0 && r != -ENOENT) {
+    return r;
+  } else if (r == 0 && mirror_mode != cls::rbd::MIRROR_MODE_DISABLED) {
+    CLS_ERR("cannot set mirror uuid while mirroring enabled");
+    return -EINVAL;
+  }
+
+  bufferlist mirror_uuid_bl;
+  mirror_uuid_bl.append(mirror_uuid);
+  r = cls_cxx_map_set_val(hctx, mirror::UUID, &mirror_uuid_bl);
+  if (r < 0) {
+    CLS_ERR("failed to set mirror uuid");
+    return r;
+  }
+  return 0;
+}
+
 /**
  * Input:
  * none
@@ -3168,6 +3255,14 @@ int mirror_mode_set(cls_method_context_t hctx, bufferlist *in,
 
   int r;
   if (enabled) {
+    std::string mirror_uuid;
+    r = mirror::uuid_get(hctx, &mirror_uuid);
+    if (r == -ENOENT) {
+      return -EINVAL;
+    } else if (r < 0) {
+      return r;
+    }
+
     bufferlist bl;
     ::encode(mirror_mode_decode, bl);
 
@@ -3188,9 +3283,13 @@ int mirror_mode_set(cls_method_context_t hctx, bufferlist *in,
       return -EBUSY;
     }
 
-    r = cls_cxx_map_remove_key(hctx, mirror::MODE);
-    if (r < 0 && r != -ENOENT) {
-      CLS_ERR("error disabling mirroring: %s", cpp_strerror(r).c_str());
+    r = remove_key(hctx, mirror::MODE);
+    if (r < 0) {
+      return r;
+    }
+
+    r = remove_key(hctx, mirror::UUID);
+    if (r < 0) {
       return r;
     }
   }
@@ -3242,6 +3341,17 @@ int mirror_peer_add(cls_method_context_t hctx, bufferlist *in,
              mirror_mode_decode == cls::rbd::MIRROR_MODE_DISABLED) {
     CLS_ERR("mirroring must be enabled on the pool");
     return -EINVAL;
+  } else if (!mirror_peer.is_valid()) {
+    CLS_ERR("mirror peer is not valid");
+    return -EINVAL;
+  }
+
+  std::string mirror_uuid;
+  r = mirror::uuid_get(hctx, &mirror_uuid);
+  if (mirror_peer.uuid == mirror_uuid) {
+    CLS_ERR("peer uuid '%s' matches pool mirroring uuid",
+            mirror_uuid.c_str());
+    return -EINVAL;
   }
 
   std::vector<cls::rbd::MirrorPeer> peers;
@@ -3626,6 +3736,11 @@ void __cls_init()
 			  old_snapshot_rename, &h_old_snapshot_rename);
 
   /* methods for the rbd_mirroring object */
+  cls_register_cxx_method(h_class, "mirror_uuid_get", CLS_METHOD_RD,
+                          mirror_uuid_get, &h_mirror_uuid_get);
+  cls_register_cxx_method(h_class, "mirror_uuid_set",
+                          CLS_METHOD_RD | CLS_METHOD_WR,
+                          mirror_uuid_set, &h_mirror_uuid_set);
   cls_register_cxx_method(h_class, "mirror_mode_get", CLS_METHOD_RD,
                           mirror_mode_get, &h_mirror_mode_get);
   cls_register_cxx_method(h_class, "mirror_mode_set",

src/cls/rbd/cls_rbd_client.cc

@@ -980,6 +980,37 @@ namespace librbd {
       return 0;
     }
 
+    int mirror_uuid_get(librados::IoCtx *ioctx, std::string *uuid) {
+      bufferlist in_bl;
+      bufferlist out_bl;
+      int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_uuid_get", in_bl,
+                          out_bl);
+      if (r < 0) {
+        return r;
+      }
+
+      try {
+        bufferlist::iterator bl_it = out_bl.begin();
+        ::decode(*uuid, bl_it);
+      } catch (const buffer::error &err) {
+        return -EBADMSG;
+      }
+      return 0;
+    }
+
+    int mirror_uuid_set(librados::IoCtx *ioctx, const std::string &uuid) {
+      bufferlist in_bl;
+      ::encode(uuid, in_bl);
+
+      bufferlist out_bl;
+      int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_uuid_set", in_bl,
+                          out_bl);
+      if (r < 0) {
+        return r;
+      }
+      return 0;
+    }
+
     int mirror_mode_get(librados::IoCtx *ioctx,
                         cls::rbd::MirrorMode *mirror_mode) {
       bufferlist in_bl;

src/cls/rbd/cls_rbd_client.h

@@ -207,6 +207,8 @@ namespace librbd {
 			  ::SnapContext *snapc);
 
     // operations on the rbd_mirroring object
+    int mirror_uuid_get(librados::IoCtx *ioctx, std::string *uuid);
+    int mirror_uuid_set(librados::IoCtx *ioctx, const std::string &uuid);
     int mirror_mode_get(librados::IoCtx *ioctx,
                         cls::rbd::MirrorMode *mirror_mode);
     int mirror_mode_set(librados::IoCtx *ioctx,

src/cls/rbd/cls_rbd_types.h

@@ -35,6 +35,10 @@ struct MirrorPeer {
   std::string client_name;
   int64_t pool_id = -1;
 
+  inline bool is_valid() const {
+    return (!uuid.empty() && !cluster_name.empty() && !client_name.empty());
+  }
+
   void encode(bufferlist &bl) const;
   void decode(bufferlist::iterator &it);
   void dump(Formatter *f) const;

src/journal/JournalMetadata.cc

@@ -591,7 +591,9 @@ void JournalMetadata::schedule_commit_task() {
 void JournalMetadata::handle_commit_position_task() {
   assert(m_timer_lock.is_locked());
   assert(m_lock.is_locked());
-  ldout(m_cct, 20) << __func__ << dendl;
+  ldout(m_cct, 20) << __func__ << ": "
+                   << "client_id=" << m_client_id << ", "
+                   << "commit_position=" << m_commit_position << dendl;
 
   librados::ObjectWriteOperation op;
   client::client_commit(&op, m_client_id, m_commit_position);