From: Jason Dillaman <dillaman@redhat.com>
Date: Fri, 6 Feb 2015 07:35:15 +0000 (-0500)
Subject: librbd: optionally blacklist clients before breaking locks
X-Git-Tag: v0.93~54^2~2
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=25d7ac2ba33ea57969813eda076c740afee2fa06;p=ceph.git

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 b18561b68104..7994e3e8fb4d 100644
--- 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 731f14dae408..3d55b5e71973 100644
--- 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 2049553cbfa5..51ec0d9f43e1 100644
--- 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)