librbd: optionally blacklist clients before breaking locks

Prior to breaking a client's lock, blacklist the clients from the
OSDs to prevent possible race conditions with data updates. Added
a new RBD configuration option to control the behavior.

Fixes: #10761
Signed-off-by: Jason Dillaman <dillaman@redhat.com>

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index b18561b681044e8e1aeec10777505d68a4bdb519..7994e3e8fb4d158934501b69398ac4949c7d3c83 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -864,6 +864,8 @@ OPTION(rbd_readahead_max_bytes, OPT_LONGLONG, 512 * 1024) // set to 0 to disable
 OPTION(rbd_readahead_disable_after_bytes, OPT_LONGLONG, 50 * 1024 * 1024) // how many bytes are read in total before readahead is disabled
 OPTION(rbd_clone_copy_on_read, OPT_BOOL, false)
 OPTION(rbd_object_map, OPT_BOOL, false) // whether to enable the RBD object map
+OPTION(rbd_blacklist_on_break_lock, OPT_BOOL, true) // whether to blacklist clients whose lock was broken
+OPTION(rbd_blacklist_expire_seconds, OPT_INT, 0) // number of seconds to blacklist - set to 0 for OSD default
 
 /*
  * The following options change the behavior for librbd's image creation methods that

diff --git a/src/librbd/ImageWatcher.cc b/src/librbd/ImageWatcher.cc
index 731f14dae408a933f25d60e164ac1fdfe5197215..3d55b5e71973cfb744f9bc225037da88a60bda46 100644 (file)
--- a/src/librbd/ImageWatcher.cc
+++ b/src/librbd/ImageWatcher.cc
@@ -184,6 +184,20 @@ int ImageWatcher::try_lock() {
       }
     }
 
+    md_config_t *conf = m_image_ctx.cct->_conf;
+    if (conf->rbd_blacklist_on_break_lock) {
+      ldout(m_image_ctx.cct, 1) << "blacklisting client: " << locker << "@"
+                               << locker_address << dendl;
+      librados::Rados rados(m_image_ctx.md_ctx);
+      r = rados.blacklist_add(locker_address,
+                             conf->rbd_blacklist_expire_seconds);
+      if (r < 0) {
+        lderr(m_image_ctx.cct) << "unable to blacklist client: "
+                              << cpp_strerror(r) << dendl;
+        return r;
+      }
+    }
+
     ldout(m_image_ctx.cct, 1) << "breaking exclusive lock: " << locker << dendl;
     r = rados::cls::lock::break_lock(&m_image_ctx.md_ctx,
                                      m_image_ctx.header_oid, RBD_LOCK_NAME,

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index 2049553cbfa509ecba91e2eb58f225a46af723fd..51ec0d9f43e1420c83fffb1f5529df5436977cea 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -2597,7 +2597,46 @@ reprotect_and_return_err:
                       << "'" << dendl;
       return -EINVAL;
     }
-    RWLock::RLocker locker(ictx->md_lock);
+
+    md_config_t *conf = ictx->cct->_conf;
+    if (conf->rbd_blacklist_on_break_lock) {
+      typedef std::map<rados::cls::lock::locker_id_t,
+                      rados::cls::lock::locker_info_t> Lockers;
+      Lockers lockers;
+      ClsLockType lock_type;
+      std::string lock_tag;
+      r = rados::cls::lock::get_lock_info(&ictx->md_ctx, ictx->header_oid,
+                                          RBD_LOCK_NAME, &lockers, &lock_type,
+                                          &lock_tag);
+      if (r < 0) {
+        lderr(ictx->cct) << "unable to retrieve lock info: " << cpp_strerror(r)
+                      << dendl;
+        return r;
+      }
+
+      std::string client_address;
+      for (Lockers::iterator it = lockers.begin();
+           it != lockers.end(); ++it) {
+        if (it->first.locker == lock_client) {
+          client_address = stringify(it->second.addr);
+          break;
+        }
+      }
+      if (client_address.empty()) {
+        return -ENOENT;
+      }
+      
+      RWLock::RLocker locker(ictx->md_lock);
+      librados::Rados rados(ictx->md_ctx);
+      r = rados.blacklist_add(client_address,
+                             conf->rbd_blacklist_expire_seconds);
+      if (r < 0) {
+        lderr(ictx->cct) << "unable to blacklist client: " << cpp_strerror(r)
+                      << dendl;
+        return r;
+      }
+    }
+
     r = rados::cls::lock::break_lock(&ictx->md_ctx, ictx->header_oid,
                                     RBD_LOCK_NAME, cookie, lock_client);
     if (r < 0)