rbd-mirror: replicate image metadata settings

Fixes: http://tracker.ceph.com/issues/16212
Signed-off-by: Mykola Golub <mgolub@mirantis.com>

src/librbd/CMakeLists.txt

@@ -64,6 +64,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

@@ -2364,43 +2364,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);
 

src/librbd/journal/Types.cc

@@ -247,6 +247,39 @@ void UpdateFeaturesEvent::dump(Formatter *f) const {
   f->dump_bool("enabled", enabled);
 }
 
+void MetadataSetEvent::encode(bufferlist& bl) const {
+  OpEventBase::encode(bl);
+  ::encode(key, bl);
+  ::encode(value, bl);
+}
+
+void MetadataSetEvent::decode(__u8 version, bufferlist::iterator& it) {
+  OpEventBase::decode(version, it);
+  ::decode(key, it);
+  ::decode(value, it);
+}
+
+void MetadataSetEvent::dump(Formatter *f) const {
+  OpEventBase::dump(f);
+  f->dump_string("key", key);
+  f->dump_string("value", value);
+}
+
+void MetadataRemoveEvent::encode(bufferlist& bl) const {
+  OpEventBase::encode(bl);
+  ::encode(key, bl);
+}
+
+void MetadataRemoveEvent::decode(__u8 version, bufferlist::iterator& it) {
+  OpEventBase::decode(version, it);
+  ::decode(key, it);
+}
+
+void MetadataRemoveEvent::dump(Formatter *f) const {
+  OpEventBase::dump(f);
+  f->dump_string("key", key);
+}
+
 void UnknownEvent::encode(bufferlist& bl) const {
   assert(false);
 }
@@ -320,6 +353,12 @@ void EventEntry::decode(bufferlist::iterator& it) {
   case EVENT_TYPE_UPDATE_FEATURES:
     event = UpdateFeaturesEvent();
     break;
+  case EVENT_TYPE_METADATA_SET:
+    event = MetadataSetEvent();
+    break;
+  case EVENT_TYPE_METADATA_REMOVE:
+    event = MetadataRemoveEvent();
+    break;
   default:
     event = UnknownEvent();
     break;
@@ -376,6 +415,12 @@ void EventEntry::generate_test_instances(std::list<EventEntry *> &o) {
 
   o.push_back(new EventEntry(UpdateFeaturesEvent()));
   o.push_back(new EventEntry(UpdateFeaturesEvent(123, 127, true)));
+
+  o.push_back(new EventEntry(MetadataSetEvent()));
+  o.push_back(new EventEntry(MetadataSetEvent(123, "key", "value")));
+
+  o.push_back(new EventEntry(MetadataRemoveEvent()));
+  o.push_back(new EventEntry(MetadataRemoveEvent(123, "key")));
 }
 
 // Journal Client
@@ -630,6 +675,12 @@ std::ostream &operator<<(std::ostream &out, const EventType &type) {
   case EVENT_TYPE_UPDATE_FEATURES:
     out << "UpdateFeatures";
     break;
+  case EVENT_TYPE_METADATA_SET:
+    out << "MetadataSet";
+    break;
+  case EVENT_TYPE_METADATA_REMOVE:
+    out << "MetadataRemove";
+    break;
   default:
     out << "Unknown (" << static_cast<uint32_t>(type) << ")";
     break;