test/librados_test_stub: support blacklisting of connections

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

src/test/librados_test_stub/LibradosTestStub.cc

@@ -3,6 +3,7 @@
 
 #include "test/librados_test_stub/LibradosTestStub.h"
 #include "include/rados/librados.hpp"
+#include "include/stringify.h"
 #include "common/ceph_argparse.h"
 #include "common/common_init.h"
 #include "common/config.h"
@@ -16,6 +17,7 @@
 #include "test/librados_test_stub/TestMemRadosClient.h"
 #include "objclass/objclass.h"
 #include "osd/osd_types.h"
+#include <arpa/inet.h>
 #include <boost/bind.hpp>
 #include <boost/shared_ptr.hpp>
 #include <deque>
@@ -1063,7 +1065,24 @@ int cls_cxx_create(cls_method_context_t hctx, bool exclusive) {
 }
 
 int cls_get_request_origin(cls_method_context_t hctx, entity_inst_t *origin) {
-  //TODO
+  librados::TestClassHandler::MethodContext *ctx =
+    reinterpret_cast<librados::TestClassHandler::MethodContext*>(hctx);
+
+  librados::TestRadosClient *rados_client =
+    ctx->io_ctx_impl->get_rados_client();
+
+  struct sockaddr_in sin;
+  sin.sin_family = AF_INET;
+  sin.sin_port = 0;
+  inet_pton(AF_INET, "127.0.0.1", &sin.sin_addr);
+
+  entity_addr_t entity_addr(entity_addr_t::TYPE_DEFAULT,
+                            rados_client->get_nonce());
+  entity_addr.in4_addr() = sin;
+
+  *origin = entity_inst_t(
+    entity_name_t::CLIENT(rados_client->get_instance_id()),
+    entity_addr);
   return 0;
 }
 

src/test/librados_test_stub/TestCluster.h

@@ -24,7 +24,7 @@ public:
     return &m_watch_notify;
   }
 
-private:
+protected:
   TestWatchNotify m_watch_notify;
 
 };

src/test/librados_test_stub/TestIoCtxImpl.cc

@@ -7,6 +7,7 @@
 #include "test/librados_test_stub/TestWatchNotify.h"
 #include "librados/AioCompletionImpl.h"
 #include "include/assert.h"
+#include "common/Finisher.h"
 #include "common/valgrind.h"
 #include "objclass/objclass.h"
 #include <boost/bind.hpp>
@@ -67,7 +68,7 @@ void TestIoCtxImpl::put() {
 }
 
 uint64_t TestIoCtxImpl::get_instance_id() const {
-  return 0;
+  return m_client->get_instance_id();
 }
 
 int64_t TestIoCtxImpl::get_id() {
@@ -131,8 +132,12 @@ int TestIoCtxImpl::aio_watch(const std::string& o, AioCompletionImpl *c,
   m_pending_ops.inc();
   c->get();
   C_AioNotify *ctx = new C_AioNotify(this, c);
-  m_client->get_watch_notify()->aio_watch(m_client, o, get_instance_id(),
-                                          handle, watch_ctx, ctx);
+  if (m_client->is_blacklisted()) {
+    m_client->get_aio_finisher()->queue(ctx, -EBLACKLISTED);
+  } else {
+    m_client->get_watch_notify()->aio_watch(m_client, o, get_instance_id(),
+                                            handle, watch_ctx, ctx);
+  }
   return 0;
 }
 
@@ -140,7 +145,11 @@ int TestIoCtxImpl::aio_unwatch(uint64_t handle, AioCompletionImpl *c) {
   m_pending_ops.inc();
   c->get();
   C_AioNotify *ctx = new C_AioNotify(this, c);
-  m_client->get_watch_notify()->aio_unwatch(m_client, handle, ctx);
+  if (m_client->is_blacklisted()) {
+    m_client->get_aio_finisher()->queue(ctx, -EBLACKLISTED);
+  } else {
+    m_client->get_watch_notify()->aio_unwatch(m_client, handle, ctx);
+  }
   return 0;
 }
 
@@ -148,6 +157,10 @@ int TestIoCtxImpl::exec(const std::string& oid, TestClassHandler *handler,
                         const char *cls, const char *method,
                         bufferlist& inbl, bufferlist* outbl,
                         const SnapContext &snapc) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   cls_method_cxx_call_t call = handler->get_method(cls, method);
   if (call == NULL) {
     return -ENOSYS;
@@ -159,11 +172,19 @@ int TestIoCtxImpl::exec(const std::string& oid, TestClassHandler *handler,
 
 int TestIoCtxImpl::list_watchers(const std::string& o,
                                  std::list<obj_watch_t> *out_watchers) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   return m_client->get_watch_notify()->list_watchers(o, out_watchers);
 }
 
 int TestIoCtxImpl::notify(const std::string& o, bufferlist& bl,
                           uint64_t timeout_ms, bufferlist *pbl) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   return m_client->get_watch_notify()->notify(m_client, o, bl, timeout_ms, pbl);
 }
 
@@ -239,6 +260,10 @@ void TestIoCtxImpl::set_snap_read(snap_t seq) {
 }
 
 int TestIoCtxImpl::tmap_update(const std::string& oid, bufferlist& cmdbl) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   // TODO: protect against concurrent tmap updates
   bufferlist tmap_header;
   std::map<string,bufferlist> tmap;
@@ -292,17 +317,29 @@ int TestIoCtxImpl::tmap_update(const std::string& oid, bufferlist& cmdbl) {
 }
 
 int TestIoCtxImpl::unwatch(uint64_t handle) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   return m_client->get_watch_notify()->unwatch(m_client, handle);
 }
 
 int TestIoCtxImpl::watch(const std::string& o, uint64_t *handle,
                          librados::WatchCtx *ctx, librados::WatchCtx2 *ctx2) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   return m_client->get_watch_notify()->watch(m_client, o, get_instance_id(),
                                              handle, ctx, ctx2);
 }
 
 int TestIoCtxImpl::execute_operation(const std::string& oid,
                                      const Operation &operation) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   TestRadosClient::Transaction transaction(m_client, oid);
   return operation(this, oid);
 }
@@ -311,13 +348,17 @@ int TestIoCtxImpl::execute_aio_operations(const std::string& oid,
                                           TestObjectOperationImpl *ops,
                                           bufferlist *pbl,
                                           const SnapContext &snapc) {
-  TestRadosClient::Transaction transaction(m_client, oid);
   int ret = 0;
-  for (ObjectOperations::iterator it = ops->ops.begin();
-       it != ops->ops.end(); ++it) {
-    ret = (*it)(this, oid, pbl, snapc);
-    if (ret < 0) {
-      break;
+  if (m_client->is_blacklisted()) {
+    ret = -EBLACKLISTED;
+  } else {
+    TestRadosClient::Transaction transaction(m_client, oid);
+    for (ObjectOperations::iterator it = ops->ops.begin();
+         it != ops->ops.end(); ++it) {
+      ret = (*it)(this, oid, pbl, snapc);
+      if (ret < 0) {
+        break;
+      }
     }
   }
   m_pending_ops.dec();

src/test/librados_test_stub/TestMemCluster.cc

@@ -23,7 +23,8 @@ TestMemCluster::Pool::Pool()
 }
 
 TestMemCluster::TestMemCluster()
-  : m_lock("TestMemCluster::m_lock") {
+  : m_lock("TestMemCluster::m_lock"),
+    m_next_nonce(static_cast<uint32_t>(reinterpret_cast<uint64_t>(this))) {
 }
 
 TestMemCluster::~TestMemCluster() {
@@ -112,6 +113,29 @@ TestMemCluster::Pool *TestMemCluster::get_pool(const std::string &pool_name) {
   return nullptr;
 }
 
+void TestMemCluster::allocate_client(uint32_t *nonce, uint64_t *global_id) {
+  Mutex::Locker locker(m_lock);
+  *nonce = m_next_nonce++;
+  *global_id = m_next_global_id++;
+}
+
+void TestMemCluster::deallocate_client(uint32_t nonce) {
+  Mutex::Locker locker(m_lock);
+  m_blacklist.erase(nonce);
+}
+
+bool TestMemCluster::is_blacklisted(uint32_t nonce) const {
+  Mutex::Locker locker(m_lock);
+  return (m_blacklist.find(nonce) != m_blacklist.end());
+}
+
+void TestMemCluster::blacklist(uint32_t nonce) {
+  m_watch_notify.blacklist(nonce);
+
+  Mutex::Locker locker(m_lock);
+  m_blacklist.insert(nonce);
+}
+
 void TestMemCluster::transaction_start(const std::string &oid) {
   Mutex::Locker locker(m_lock);
   while (m_transactions.count(oid)) {

src/test/librados_test_stub/TestMemCluster.h

@@ -15,6 +15,7 @@
 #include <boost/shared_ptr.hpp>
 #include <list>
 #include <map>
+#include <set>
 #include <string>
 
 namespace librados {
@@ -77,18 +78,30 @@ public:
   Pool *get_pool(int64_t pool_id);
   Pool *get_pool(const std::string &pool_name);
 
+  void allocate_client(uint32_t *nonce, uint64_t *global_id);
+  void deallocate_client(uint32_t nonce);
+
+  bool is_blacklisted(uint32_t nonce) const;
+  void blacklist(uint32_t nonce);
+
   void transaction_start(const std::string &oid);
   void transaction_finish(const std::string &oid);
 
 private:
 
   typedef std::map<std::string, Pool*>		Pools;
+  typedef std::set<uint32_t> Blacklist;
 
   mutable Mutex m_lock;
 
   Pools	m_pools;
   int64_t m_pool_id = 0;
 
+  uint32_t m_next_nonce;
+  uint64_t m_next_global_id = 1234;
+
+  Blacklist m_blacklist;
+
   Cond m_transaction_cond;
   std::set<std::string> m_transactions;
 

src/test/librados_test_stub/TestMemIoCtxImpl.cc

@@ -57,6 +57,8 @@ int TestMemIoCtxImpl::append(const std::string& oid, const bufferlist &bl,
                              const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -71,6 +73,10 @@ int TestMemIoCtxImpl::append(const std::string& oid, const bufferlist &bl,
 }
 
 int TestMemIoCtxImpl::assert_exists(const std::string &oid) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   RWLock::RLocker l(m_pool->file_lock);
   TestMemCluster::SharedFile file = get_file(oid, false, get_snap_context());
   if (file == NULL) {
@@ -82,6 +88,8 @@ int TestMemIoCtxImpl::assert_exists(const std::string &oid) {
 int TestMemIoCtxImpl::create(const std::string& oid, bool exclusive) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   RWLock::WLocker l(m_pool->file_lock);
@@ -90,6 +98,10 @@ int TestMemIoCtxImpl::create(const std::string& oid, bool exclusive) {
 }
 
 int TestMemIoCtxImpl::list_snaps(const std::string& oid, snap_set_t *out_snaps) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   out_snaps->seq = 0;
   out_snaps->clones.clear();
 
@@ -162,6 +174,8 @@ int TestMemIoCtxImpl::omap_get_vals(const std::string& oid,
                                     std::map<std::string, bufferlist> *out_vals) {
   if (out_vals == NULL) {
     return -EINVAL;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -202,6 +216,8 @@ int TestMemIoCtxImpl::omap_rm_keys(const std::string& oid,
                                    const std::set<std::string>& keys) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -225,6 +241,8 @@ int TestMemIoCtxImpl::omap_set(const std::string& oid,
                                const std::map<std::string, bufferlist> &map) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -249,6 +267,10 @@ int TestMemIoCtxImpl::omap_set(const std::string& oid,
 
 int TestMemIoCtxImpl::read(const std::string& oid, size_t len, uint64_t off,
                            bufferlist *bl) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   TestMemCluster::SharedFile file;
   {
     RWLock::RLocker l(m_pool->file_lock);
@@ -274,6 +296,8 @@ int TestMemIoCtxImpl::read(const std::string& oid, size_t len, uint64_t off,
 int TestMemIoCtxImpl::remove(const std::string& oid, const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   RWLock::WLocker l(m_pool->file_lock);
@@ -296,6 +320,10 @@ int TestMemIoCtxImpl::remove(const std::string& oid, const SnapContext &snapc) {
 }
 
 int TestMemIoCtxImpl::selfmanaged_snap_create(uint64_t *snapid) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   RWLock::WLocker l(m_pool->file_lock);
   *snapid = ++m_pool->snap_id;
   m_pool->snap_seqs.insert(*snapid);
@@ -303,6 +331,10 @@ int TestMemIoCtxImpl::selfmanaged_snap_create(uint64_t *snapid) {
 }
 
 int TestMemIoCtxImpl::selfmanaged_snap_remove(uint64_t snapid) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   RWLock::WLocker l(m_pool->file_lock);
   TestMemCluster::SnapSeqs::iterator it =
     m_pool->snap_seqs.find(snapid);
@@ -317,6 +349,10 @@ int TestMemIoCtxImpl::selfmanaged_snap_remove(uint64_t snapid) {
 
 int TestMemIoCtxImpl::selfmanaged_snap_rollback(const std::string& oid,
                                                 uint64_t snapid) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   RWLock::WLocker l(m_pool->file_lock);
 
   TestMemCluster::SharedFile file;
@@ -363,6 +399,10 @@ int TestMemIoCtxImpl::sparse_read(const std::string& oid, uint64_t off,
                                   uint64_t len,
                                   std::map<uint64_t,uint64_t> *m,
                                   bufferlist *data_bl) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   // TODO verify correctness
   TestMemCluster::SharedFile file;
   {
@@ -391,6 +431,10 @@ int TestMemIoCtxImpl::sparse_read(const std::string& oid, uint64_t off,
 
 int TestMemIoCtxImpl::stat(const std::string& oid, uint64_t *psize,
                            time_t *pmtime) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   TestMemCluster::SharedFile file;
   {
     RWLock::RLocker l(m_pool->file_lock);
@@ -414,6 +458,8 @@ int TestMemIoCtxImpl::truncate(const std::string& oid, uint64_t size,
                                const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -450,6 +496,8 @@ int TestMemIoCtxImpl::write(const std::string& oid, bufferlist& bl, size_t len,
                             uint64_t off, const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -475,6 +523,8 @@ int TestMemIoCtxImpl::write_full(const std::string& oid, bufferlist& bl,
                                  const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
   TestMemCluster::SharedFile file;
@@ -503,8 +553,11 @@ int TestMemIoCtxImpl::writesame(const std::string& oid, bufferlist& bl, size_t l
                                 uint64_t off, const SnapContext &snapc) {
   if (get_snap_read() != CEPH_NOSNAP) {
     return -EROFS;
+  } else if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
   }
 
+
   if (len == 0 || (len % bl.length())) {
     return -EINVAL;
   }
@@ -534,6 +587,10 @@ int TestMemIoCtxImpl::writesame(const std::string& oid, bufferlist& bl, size_t l
 
 int TestMemIoCtxImpl::xattr_get(const std::string& oid,
                                 std::map<std::string, bufferlist>* attrset) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   TestMemCluster::SharedFile file;
   RWLock::RLocker l(m_pool->file_lock);
   TestMemCluster::FileXAttrs::iterator it = m_pool->file_xattrs.find(oid);
@@ -546,12 +603,20 @@ int TestMemIoCtxImpl::xattr_get(const std::string& oid,
 
 int TestMemIoCtxImpl::xattr_set(const std::string& oid, const std::string &name,
                                 bufferlist& bl) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   RWLock::WLocker l(m_pool->file_lock);
   m_pool->file_xattrs[oid][name] = bl;
   return 0;
 }
 
 int TestMemIoCtxImpl::zero(const std::string& oid, uint64_t off, uint64_t len) {
+  if (m_client->is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
   bool truncate_redirect = false;
   TestMemCluster::SharedFile file;
   {

src/test/librados_test_stub/TestMemRadosClient.cc

@@ -5,6 +5,7 @@
 #include "test/librados_test_stub/TestMemCluster.h"
 #include "test/librados_test_stub/TestMemIoCtxImpl.h"
 #include <errno.h>
+#include <sstream>
 
 namespace librados {
 
@@ -12,6 +13,11 @@ TestMemRadosClient::TestMemRadosClient(CephContext *cct,
                                        TestMemCluster *test_mem_cluster)
   : TestRadosClient(cct, test_mem_cluster->get_watch_notify()),
     m_mem_cluster(test_mem_cluster) {
+  m_mem_cluster->allocate_client(&m_nonce, &m_global_id);
+}
+
+TestMemRadosClient::~TestMemRadosClient() {
+  m_mem_cluster->deallocate_client(m_nonce);
 }
 
 TestIoCtxImpl *TestMemRadosClient::create_ioctx(int64_t pool_id,
@@ -36,10 +42,16 @@ void TestMemRadosClient::object_list(int64_t pool_id,
 }
 
 int TestMemRadosClient::pool_create(const std::string &pool_name) {
+  if (is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
   return m_mem_cluster->pool_create(pool_name);
 }
 
 int TestMemRadosClient::pool_delete(const std::string &pool_name) {
+  if (is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
   return m_mem_cluster->pool_delete(pool_name);
 }
 
@@ -66,8 +78,30 @@ int TestMemRadosClient::watch_flush() {
   return 0;
 }
 
+bool TestMemRadosClient::is_blacklisted() const {
+  return m_mem_cluster->is_blacklisted(m_nonce);
+}
+
 int TestMemRadosClient::blacklist_add(const std::string& client_address,
 				      uint32_t expire_seconds) {
+  if (is_blacklisted()) {
+    return -EBLACKLISTED;
+  }
+
+  // extract the nonce to use as a unique key to the client
+  auto idx = client_address.find("/");
+  if (idx == std::string::npos || idx + 1 >= client_address.size()) {
+    return -EINVAL;
+  }
+
+  std::stringstream nonce_ss(client_address.substr(idx + 1));
+  uint32_t nonce;
+  nonce_ss >> nonce;
+  if (!nonce_ss) {
+    return -EINVAL;
+  }
+
+  m_mem_cluster->blacklist(nonce);
   return 0;
 }
 

src/test/librados_test_stub/TestMemRadosClient.h

@@ -17,10 +17,18 @@ class TestMemCluster;
 class TestMemRadosClient : public TestRadosClient {
 public:
   TestMemRadosClient(CephContext *cct, TestMemCluster *test_mem_cluster);
+  ~TestMemRadosClient() override;
 
   TestIoCtxImpl *create_ioctx(int64_t pool_id,
                                       const std::string &pool_name) override;
 
+  uint32_t get_nonce() override {
+    return m_nonce;
+  }
+  uint64_t get_instance_id() override {
+    return m_global_id;
+  }
+
   void object_list(int64_t pool_id,
                    std::list<librados::TestRadosClient::Object> *list) override;
 
@@ -33,6 +41,7 @@ public:
 
   int watch_flush() override;
 
+  bool is_blacklisted() const override;
   int blacklist_add(const std::string& client_address,
                     uint32_t expire_seconds) override;
 protected:
@@ -46,6 +55,8 @@ protected:
 
 private:
   TestMemCluster *m_mem_cluster;
+  uint32_t m_nonce;
+  uint64_t m_global_id;
 
 };
 

src/test/librados_test_stub/TestRadosClient.cc

@@ -128,10 +128,6 @@ CephContext *TestRadosClient::cct() {
   return m_cct;
 }
 
-uint64_t TestRadosClient::get_instance_id() {
-  return 0;
-}
-
 int TestRadosClient::connect() {
   return 0;
 }

src/test/librados_test_stub/TestRadosClient.h

@@ -59,7 +59,8 @@ public:
 
   virtual CephContext *cct();
 
-  virtual uint64_t get_instance_id();
+  virtual uint32_t get_nonce() = 0;
+  virtual uint64_t get_instance_id() = 0;
 
   virtual int connect();
   virtual void shutdown();
@@ -85,6 +86,7 @@ public:
   virtual int aio_watch_flush(AioCompletionImpl *c);
   virtual int watch_flush() = 0;
 
+  virtual bool is_blacklisted() const = 0;
   virtual int blacklist_add(const std::string& client_address,
 			    uint32_t expire_seconds) = 0;
 

src/test/librados_test_stub/TestWatchNotify.cc

@@ -47,7 +47,7 @@ int TestWatchNotify::list_watchers(const std::string& o,
          watcher->watch_handles.begin();
        it != watcher->watch_handles.end(); ++it) {
     obj_watch_t obj;
-    strcpy(obj.addr, "-");
+    strcpy(obj.addr, it->second.addr.c_str());
     obj.watcher_id = static_cast<int64_t>(it->second.gid);
     obj.cookie = it->second.handle;
     obj.timeout_seconds = 30;
@@ -139,6 +139,8 @@ int TestWatchNotify::watch(TestRadosClient *rados_client,
 
   WatchHandle watch_handle;
   watch_handle.rados_client = rados_client;
+  watch_handle.addr = "127.0.0.1:0/" + stringify(rados_client->get_nonce());
+  watch_handle.nonce = rados_client->get_nonce();
   watch_handle.gid = gid;
   watch_handle.handle = ++m_handle;
   watch_handle.watch_ctx = ctx;
@@ -312,4 +314,26 @@ void TestWatchNotify::finish_notify(TestRadosClient *rados_client,
   }
 }
 
+void TestWatchNotify::blacklist(uint32_t nonce) {
+  Mutex::Locker locker(m_lock);
+
+  for (auto file_it = m_file_watchers.begin();
+       file_it != m_file_watchers.end(); ) {
+    auto &watcher = file_it->second;
+    for (auto w_it = watcher->watch_handles.begin();
+         w_it != watcher->watch_handles.end();) {
+      if (w_it->second.nonce == nonce) {
+        w_it = watcher->watch_handles.erase(w_it);
+      } else {
+        ++w_it;
+      }
+    }
+    if (watcher->watch_handles.empty() && watcher->notify_handles.empty()) {
+        file_it = m_file_watchers.erase(file_it);
+    } else {
+      ++file_it;
+    }
+  }
+}
+
 } // namespace librados

src/test/librados_test_stub/TestWatchNotify.h

@@ -37,6 +37,8 @@ public:
 
   struct WatchHandle {
     TestRadosClient *rados_client = nullptr;
+    std::string addr;
+    uint32_t nonce;
     uint64_t gid;
     uint64_t handle;
     librados::WatchCtx* watch_ctx;
@@ -77,6 +79,8 @@ public:
             librados::WatchCtx2 *ctx2);
   int unwatch(TestRadosClient *rados_client, uint64_t handle);
 
+  void blacklist(uint32_t nonce);
+
 private:
 
   typedef std::map<std::string, SharedWatcher> FileWatchers;

src/test/librbd/test_librbd.cc

@@ -221,12 +221,6 @@ public:
     return value == "true";
   }
 
-  bool is_librados_test_stub() {
-    std::string fsid;
-    EXPECT_EQ(0, _rados.cluster_fsid(&fsid));
-    return fsid == "00000000-1111-2222-3333-444444444444";
-  }
-
   void validate_object_map(rbd_image_t image, bool *passed) {
     uint64_t flags;
     ASSERT_EQ(0, rbd_get_flags(image, &flags));
@@ -5393,7 +5387,6 @@ TEST_F(TestLibRBD, ExclusiveLock)
 TEST_F(TestLibRBD, BreakLock)
 {
   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
-  REQUIRE(!is_librados_test_stub());
 
   static char buf[10];
 

src/test/rbd_mirror/test_LeaderWatcher.cc

@@ -209,8 +209,6 @@ TEST_F(TestLeaderWatcher, ListenerError)
 
 TEST_F(TestLeaderWatcher, Two)
 {
-  REQUIRE(!is_librados_test_stub());
-
   Listener listener1;
   LeaderWatcher<> leader_watcher1(m_threads, create_connection(), &listener1);
 
@@ -249,8 +247,6 @@ TEST_F(TestLeaderWatcher, Two)
 
 TEST_F(TestLeaderWatcher, Break)
 {
-  REQUIRE(!is_librados_test_stub());
-
   Listener listener1, listener2;
   LeaderWatcher<> leader_watcher1(m_threads,
                                   create_connection(true /* no heartbeats */),
@@ -277,8 +273,6 @@ TEST_F(TestLeaderWatcher, Break)
 
 TEST_F(TestLeaderWatcher, Stress)
 {
-  REQUIRE(!is_librados_test_stub());
-
   const int WATCHERS_COUNT = 20;
   std::list<LeaderWatcher<> *> leader_watchers;
   Listener listener;