Merge pull request #11168 from trociny/wip-16212

rbd-mirror: replicate image metadata settings

Reviewed-by: Jason Dillaman <dillaman@redhat.com>

src/cls/rbd/cls_rbd_client.cc

@@ -1013,22 +1013,40 @@ namespace librbd {
       rados_op->exec("rbd", "object_map_snap_remove", in);
     }
 
+    void metadata_set(librados::ObjectWriteOperation *op,
+                     const map<string, bufferlist> &data)
+    {
+      bufferlist bl;
+      ::encode(data, bl);
+
+      op->exec("rbd", "metadata_set", bl);
+    }
+
     int metadata_set(librados::IoCtx *ioctx, const std::string &oid,
                      const map<string, bufferlist> &data)
     {
-      bufferlist in;
-      ::encode(data, in);
-      bufferlist out;
-      return ioctx->exec(oid, "rbd", "metadata_set", in, out);
+      librados::ObjectWriteOperation op;
+      metadata_set(&op, data);
+
+      return ioctx->operate(oid, &op);
+    }
+
+    void metadata_remove(librados::ObjectWriteOperation *op,
+                         const std::string &key)
+    {
+      bufferlist bl;
+      ::encode(key, bl);
+
+      op->exec("rbd", "metadata_remove", bl);
     }
 
     int metadata_remove(librados::IoCtx *ioctx, const std::string &oid,
-                        const std::string &key)
+                     const std::string &key)
     {
-      bufferlist in;
-      ::encode(key, in);
-      bufferlist out;
-      return ioctx->exec(oid, "rbd", "metadata_remove", in, out);
+      librados::ObjectWriteOperation op;
+      metadata_remove(&op, key);
+
+      return ioctx->operate(oid, &op);
     }
 
     int metadata_list(librados::IoCtx *ioctx, const std::string &oid,

src/cls/rbd/cls_rbd_client.h

@@ -153,8 +153,12 @@ namespace librbd {
                              const std::string &start, uint64_t max_return);
     int metadata_list_finish(bufferlist::iterator *it,
                              std::map<std::string, bufferlist> *pairs);
+    void metadata_set(librados::ObjectWriteOperation *op,
+                      const map<std::string, bufferlist> &data);
     int metadata_set(librados::IoCtx *ioctx, const std::string &oid,
                      const map<std::string, bufferlist> &data);
+    void metadata_remove(librados::ObjectWriteOperation *op,
+                         const std::string &key);
     int metadata_remove(librados::IoCtx *ioctx, const std::string &oid,
                         const std::string &key);
     int metadata_get(librados::IoCtx *ioctx, const std::string &oid,

src/librbd/CMakeLists.txt

@@ -65,6 +65,8 @@ set(librbd_internal_srcs
   operation/DisableFeaturesRequest.cc
   operation/EnableFeaturesRequest.cc
   operation/FlattenRequest.cc
+  operation/MetadataRemoveRequest.cc
+  operation/MetadataSetRequest.cc
   operation/ObjectMapIterate.cc
   operation/RebuildObjectMapRequest.cc
   operation/RenameRequest.cc

src/librbd/Operations.cc

@@ -15,8 +15,10 @@
 #include "librbd/operation/DisableFeaturesRequest.h"
 #include "librbd/operation/EnableFeaturesRequest.h"
 #include "librbd/operation/FlattenRequest.h"
-#include "librbd/operation/RebuildObjectMapRequest.h"
+#include "librbd/operation/MetadataRemoveRequest.h"
+#include "librbd/operation/MetadataSetRequest.h"
 #include "librbd/operation/ObjectMapIterate.h"
+#include "librbd/operation/RebuildObjectMapRequest.h"
 #include "librbd/operation/RenameRequest.h"
 #include "librbd/operation/ResizeRequest.h"
 #include "librbd/operation/SnapshotCreateRequest.h"
@@ -1296,6 +1298,123 @@ void Operations<I>::execute_update_features(uint64_t features, bool enabled,
   }
 }
 
+template <typename I>
+int Operations<I>::metadata_set(const std::string &key,
+                                const std::string &value) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": key=" << key << ", value="
+                << value << dendl;
+
+  string start = m_image_ctx.METADATA_CONF_PREFIX;
+  size_t conf_prefix_len = start.size();
+
+  if (key.size() > conf_prefix_len && !key.compare(0, conf_prefix_len, start)) {
+    string subkey = key.substr(conf_prefix_len, key.size() - conf_prefix_len);
+    int r = cct->_conf->set_val(subkey.c_str(), value);
+    if (r < 0) {
+      return r;
+    }
+  }
+
+  int r = m_image_ctx.state->refresh_if_required();
+  if (r < 0) {
+    return r;
+  }
+
+  if (m_image_ctx.read_only) {
+    return -EROFS;
+  }
+
+  {
+    RWLock::RLocker owner_lock(m_image_ctx.owner_lock);
+    C_SaferCond metadata_ctx;
+
+    if (m_image_ctx.exclusive_lock != nullptr &&
+	!m_image_ctx.exclusive_lock->is_lock_owner()) {
+      C_SaferCond lock_ctx;
+
+      m_image_ctx.exclusive_lock->request_lock(&lock_ctx);
+      r = lock_ctx.wait();
+      if (r < 0) {
+	return r;
+      }
+    }
+
+    execute_metadata_set(key, value, &metadata_ctx);
+    r = metadata_ctx.wait();
+  }
+
+  return r;
+}
+
+template <typename I>
+void Operations<I>::execute_metadata_set(const std::string &key,
+					const std::string &value,
+					Context *on_finish) {
+  assert(m_image_ctx.owner_lock.is_locked());
+
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": key=" << key << ", value="
+                << value << dendl;
+
+  operation::MetadataSetRequest<I> *request =
+    new operation::MetadataSetRequest<I>(m_image_ctx, on_finish, key, value);
+  request->send();
+}
+
+template <typename I>
+int Operations<I>::metadata_remove(const std::string &key) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": key=" << key << dendl;
+
+  if (m_image_ctx.read_only) {
+    return -EROFS;
+  }
+
+  int r = m_image_ctx.state->refresh_if_required();
+  if (r < 0) {
+    return r;
+  }
+
+  if (m_image_ctx.read_only) {
+    return -EROFS;
+  }
+
+  {
+    RWLock::RLocker owner_lock(m_image_ctx.owner_lock);
+    C_SaferCond metadata_ctx;
+
+    if (m_image_ctx.exclusive_lock != nullptr &&
+        !m_image_ctx.exclusive_lock->is_lock_owner()) {
+      C_SaferCond lock_ctx;
+
+      m_image_ctx.exclusive_lock->request_lock(&lock_ctx);
+      r = lock_ctx.wait();
+      if (r < 0) {
+        return r;
+      }
+    }
+
+    execute_metadata_remove(key, &metadata_ctx);
+    r = metadata_ctx.wait();
+  }
+
+  return r;
+}
+
+template <typename I>
+void Operations<I>::execute_metadata_remove(const std::string &key,
+                                           Context *on_finish) {
+  assert(m_image_ctx.owner_lock.is_locked());
+
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 5) << this << " " << __func__ << ": key=" << key << dendl;
+
+  operation::MetadataRemoveRequest<I> *request =
+    new operation::MetadataRemoveRequest<I>(m_image_ctx, on_finish, key);
+  request->send();
+}
+
 template <typename I>
 int Operations<I>::prepare_image_update() {
   assert(m_image_ctx.owner_lock.is_locked() &&

src/librbd/Operations.h

@@ -74,6 +74,13 @@ class Operations {
   void execute_update_features(uint64_t features, bool enabled,
                                Context *on_finish, uint64_t journal_op_tid);
 
+  int metadata_set(const std::string &key, const std::string &value);
+  void execute_metadata_set(const std::string &key, const std::string &value,
+                            Context *on_finish);
+
+  int metadata_remove(const std::string &key);
+  void execute_metadata_remove(const std::string &key, Context *on_finish);
+
   int prepare_image_update();
 
 private:

src/librbd/internal.cc

@@ -2365,43 +2365,6 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force,
     return cls_client::metadata_get(&ictx->md_ctx, ictx->header_oid, key, value);
   }
 
-  int metadata_set(ImageCtx *ictx, const string &key, const string &value)
-  {
-    CephContext *cct = ictx->cct;
-    string start = ictx->METADATA_CONF_PREFIX;
-    size_t conf_prefix_len = start.size();
-
-    if(key.size() > conf_prefix_len && !key.compare(0,conf_prefix_len,start)) {
-      string subkey = key.substr(conf_prefix_len, key.size()-conf_prefix_len);
-      int r = cct->_conf->set_val(subkey.c_str(), value);
-      if (r < 0)
-        return r;
-    }
-    ldout(cct, 20) << "metadata_set " << ictx << " key=" << key << " value=" << value << dendl;
-
-    int r = ictx->state->refresh_if_required();
-    if (r < 0) {
-      return r;
-    }
-
-    map<string, bufferlist> data;
-    data[key].append(value);
-    return cls_client::metadata_set(&ictx->md_ctx, ictx->header_oid, data);
-  }
-
-  int metadata_remove(ImageCtx *ictx, const string &key)
-  {
-    CephContext *cct = ictx->cct;
-    ldout(cct, 20) << "metadata_remove " << ictx << " key=" << key << dendl;
-
-    int r = ictx->state->refresh_if_required();
-    if (r < 0) {
-      return r;
-    }
-
-    return cls_client::metadata_remove(&ictx->md_ctx, ictx->header_oid, key);
-  }
-
   int metadata_list(ImageCtx *ictx, const string &start, uint64_t max, map<string, bufferlist> *pairs)
   {
     CephContext *cct = ictx->cct;

src/librbd/internal.h

@@ -195,8 +195,6 @@ namespace librbd {
   int poll_io_events(ImageCtx *ictx, AioCompletion **comps, int numcomp);
   int metadata_list(ImageCtx *ictx, const string &last, uint64_t max, map<string, bufferlist> *pairs);
   int metadata_get(ImageCtx *ictx, const std::string &key, std::string *value);
-  int metadata_set(ImageCtx *ictx, const std::string &key, const std::string &value);
-  int metadata_remove(ImageCtx *ictx, const std::string &key);
 
   int mirror_mode_get(IoCtx& io_ctx, rbd_mirror_mode_t *mirror_mode);
   int mirror_mode_set(IoCtx& io_ctx, rbd_mirror_mode_t mirror_mode);

src/librbd/journal/Replay.cc

@@ -94,6 +94,15 @@ struct ExecuteOp : public Context {
 						  on_op_complete, event.op_tid);
   }
 
+  void execute(const journal::MetadataSetEvent &_) {
+    image_ctx.operations->execute_metadata_set(event.key, event.value,
+                                               on_op_complete);
+  }
+
+  void execute(const journal::MetadataRemoveEvent &_) {
+    image_ctx.operations->execute_metadata_remove(event.key, on_op_complete);
+  }
+
   virtual void finish(int r) override {
     CephContext *cct = image_ctx.cct;
     if (r < 0) {
@@ -688,6 +697,51 @@ void Replay<I>::handle_event(const journal::UpdateFeaturesEvent &event,
   op_event->on_start_ready = on_ready;
 }
 
+template <typename I>
+void Replay<I>::handle_event(const journal::MetadataSetEvent &event,
+			     Context *on_ready, Context *on_safe) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << ": Metadata set event" << dendl;
+
+  Mutex::Locker locker(m_lock);
+  OpEvent *op_event;
+  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
+                                                       on_safe, &op_event);
+  if (on_op_complete == nullptr) {
+    return;
+  }
+
+  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
+    m_image_ctx, new ExecuteOp<I, journal::MetadataSetEvent>(
+      m_image_ctx, event, on_op_complete));
+
+  on_ready->complete(0);
+}
+
+template <typename I>
+void Replay<I>::handle_event(const journal::MetadataRemoveEvent &event,
+			     Context *on_ready, Context *on_safe) {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << ": Metadata remove event" << dendl;
+
+  Mutex::Locker locker(m_lock);
+  OpEvent *op_event;
+  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
+                                                       on_safe, &op_event);
+  if (on_op_complete == nullptr) {
+    return;
+  }
+
+  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
+    m_image_ctx, new ExecuteOp<I, journal::MetadataRemoveEvent>(
+      m_image_ctx, event, on_op_complete));
+
+  // ignore errors caused due to replay
+  op_event->ignore_error_codes = {-ENOENT};
+
+  on_ready->complete(0);
+}
+
 template <typename I>
 void Replay<I>::handle_event(const journal::UnknownEvent &event,
 			     Context *on_ready, Context *on_safe) {

src/librbd/journal/Replay.h

@@ -161,6 +161,10 @@ class Replay {
 		    Context *on_safe);
   void handle_event(const UpdateFeaturesEvent &event, Context *on_ready,
                     Context *on_safe);
+  void handle_event(const MetadataSetEvent &event, Context *on_ready,
+                    Context *on_safe);
+  void handle_event(const MetadataRemoveEvent &event, Context *on_ready,
+                    Context *on_safe);
   void handle_event(const UnknownEvent &event, Context *on_ready,
                     Context *on_safe);