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 e15db05)

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

(cherry picked from commit 7e30b63)

src/librbd/ExclusiveLock.cc

@@ -250,6 +250,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());

src/librbd/ExclusiveLock.h

@@ -43,6 +43,8 @@ class ExclusiveLock {
 
   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);

src/librbd/internal.cc

@@ -2099,9 +2099,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;
   }
 

src/test/pybind/test_rbd.py

@@ -1186,6 +1186,9 @@ def test_break_lock(self):
                     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,