librbd: is_exclusive_lock_owner API should ping OSD

This is required to detect if a peer has been silently blacklisted
and is therefore no longer the lock owner.

Fixes: http://tracker.ceph.com/issues/19287
Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit e15db05960a284bdf3701256722299d553cfd5aa)

Conflicts:
	src/librbd/ManagedLock.[h|cc]: logic moved to ExclusiveLock

diff --git a/src/librbd/ExclusiveLock.cc b/src/librbd/ExclusiveLock.cc
index a268f20b87f27793c4fc9d8d8c6947dfce80108e..3c77066a3a6d6a6d0ef1d381de3ca017f8f71d44 100644 (file)
--- a/src/librbd/ExclusiveLock.cc
+++ b/src/librbd/ExclusiveLock.cc
@@ -253,6 +253,44 @@ void ExclusiveLock<I>::handle_peer_notification(int r) {
   handle_acquire_lock(r);
 }
 
+template <typename I>
+int ExclusiveLock<I>::assert_header_locked() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 10) << this << " " << __func__ << dendl;
+
+  librados::ObjectReadOperation op;
+  {
+    Mutex::Locker locker(m_lock);
+    rados::cls::lock::assert_locked(&op, RBD_LOCK_NAME, LOCK_EXCLUSIVE,
+                                    m_cookie, WATCHER_LOCK_TAG);
+  }
+
+  int r = m_image_ctx.md_ctx.operate(m_image_ctx.header_oid, &op, nullptr);
+  if (r < 0) {
+    if (r == -EBLACKLISTED) {
+      ldout(cct, 5) << this << " " << __func__ << ": "
+                   << "client is not lock owner -- client blacklisted"
+                    << dendl;
+    } else if (r == -ENOENT) {
+      ldout(cct, 5) << this << " " << __func__ << ": "
+                   << "client is not lock owner -- no lock detected"
+                    << dendl;
+    } else if (r == -EBUSY) {
+      ldout(cct, 5) << this << " " << __func__ << ": "
+                   << "client is not lock owner -- owned by different client"
+                    << dendl;
+    } else {
+      lderr(cct) << this << " " << __func__ << ": "
+                << "failed to verify lock ownership: " << cpp_strerror(r)
+                 << dendl;
+    }
+
+    return r;
+  }
+
+  return 0;
+}
+
 template <typename I>
 std::string ExclusiveLock<I>::encode_lock_cookie() const {
   assert(m_lock.is_locked());

diff --git a/src/librbd/ExclusiveLock.h b/src/librbd/ExclusiveLock.h
index 3048c2bdbcbd157a6427991481fb4547dfdda13d..545013cc15394861e5d35cab0e186957fe8211be 100644 (file)
--- a/src/librbd/ExclusiveLock.h
+++ b/src/librbd/ExclusiveLock.h
@@ -43,6 +43,8 @@ public:
 
   void reacquire_lock(Context *on_reacquired = nullptr);
 
+  int assert_header_locked();
+
   void handle_peer_notification(int r);
 
   static bool decode_lock_cookie(const std::string &cookie, uint64_t *handle);

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index 8512e6f4999cc74f479ebab77147bfa2368a3063..3681b4f73a8ecf692a7edfac4f3ddbaa46fe5345 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1410,9 +1410,22 @@ int mirror_image_disable_internal(ImageCtx *ictx, bool force,
 
   int is_exclusive_lock_owner(ImageCtx *ictx, bool *is_owner)
   {
-    RWLock::RLocker l(ictx->owner_lock);
-    *is_owner = (ictx->exclusive_lock != nullptr &&
-                ictx->exclusive_lock->is_lock_owner());
+    *is_owner = false;
+
+    RWLock::RLocker owner_locker(ictx->owner_lock);
+    if (ictx->exclusive_lock == nullptr ||
+        !ictx->exclusive_lock->is_lock_owner()) {
+      return 0;
+    }
+
+    // might have been blacklisted by peer -- ensure we still own
+    // the lock by pinging the OSD
+    int r = ictx->exclusive_lock->assert_header_locked();
+    if (r < 0) {
+      return r;
+    }
+
+    *is_owner = true;
     return 0;
   }
 

diff --git a/src/test/pybind/test_rbd.py b/src/test/pybind/test_rbd.py
index dec3aec46f067a3b7d92aa1b1a3662f53301dd13..4c93e9775a9b2946aaadb5ba1716307de7d89385 100644 (file)
--- a/src/test/pybind/test_rbd.py
+++ b/src/test/pybind/test_rbd.py
@@ -1269,6 +1269,9 @@ class TestExclusiveLock(object):
                     image.lock_break(RBD_LOCK_MODE_EXCLUSIVE,
                                      lock_owners[0]['owner'])
 
+                    assert_raises(ConnectionShutdown,
+                                  blacklist_image.is_exclusive_lock_owner)
+
                     blacklist_rados.wait_for_latest_osdmap()
                     data = rand_data(256)
                     assert_raises(ConnectionShutdown,