rgw: Add cache introspection and manipulation

Provide the ability to examine and delete elements from the cache.

Fixes: http://tracker.ceph.com/issues/22603
Fixes: http://tracker.ceph.com/issues/22604
Signed-off-by: Adam C. Emerson <aemerson@redhat.com>

src/rgw/rgw_cache.cc

@@ -8,7 +8,7 @@
 #define dout_subsys ceph_subsys_rgw
 
 
-int ObjectCache::get(string& name, ObjectCacheInfo& info, uint32_t mask, rgw_cache_entry_info *cache_info)
+int ObjectCache::get(const string& name, ObjectCacheInfo& info, uint32_t mask, rgw_cache_entry_info *cache_info)
 {
   RWLock::RLocker l(lock);
 
@@ -114,7 +114,7 @@ bool ObjectCache::chain_cache_entry(std::initializer_list<rgw_cache_entry_info*>
   return true;
 }
 
-void ObjectCache::put(string& name, ObjectCacheInfo& info, rgw_cache_entry_info *cache_info)
+void ObjectCache::put(const string& name, ObjectCacheInfo& info, rgw_cache_entry_info *cache_info)
 {
   RWLock::WLocker l(lock);
 
@@ -187,17 +187,17 @@ void ObjectCache::put(string& name, ObjectCacheInfo& info, rgw_cache_entry_info
     target.version = info.version;
 }
 
-void ObjectCache::remove(string& name)
+bool ObjectCache::remove(const string& name)
 {
   RWLock::WLocker l(lock);
 
   if (!enabled) {
-    return;
+    return false;
   }
 
   auto iter = cache_map.find(name);
   if (iter == cache_map.end())
-    return;
+    return false;
 
   ldout(cct, 10) << "removing " << name << " from cache" << dendl;
   ObjectCacheEntry& entry = iter->second;
@@ -208,9 +208,10 @@ void ObjectCache::remove(string& name)
 
   remove_lru(name, iter->second.lru_iter);
   cache_map.erase(iter);
+  return true;
 }
 
-void ObjectCache::touch_lru(string& name, ObjectCacheEntry& entry,
+void ObjectCache::touch_lru(const string& name, ObjectCacheEntry& entry,
 			    std::list<string>::iterator& lru_iter)
 {
   while (lru_size > (size_t)cct->_conf->rgw_cache_lru_size) {
@@ -250,7 +251,7 @@ void ObjectCache::touch_lru(string& name, ObjectCacheEntry& entry,
   entry.lru_promotion_ts = lru_counter;
 }
 
-void ObjectCache::remove_lru(string& name,
+void ObjectCache::remove_lru(const string& name,
 			     std::list<string>::iterator& lru_iter)
 {
   if (lru_iter == lru.end())

src/rgw/rgw_cache.h

@@ -150,17 +150,36 @@ class ObjectCache {
   bool enabled;
   ceph::timespan expiry;
 
-  void touch_lru(string& name, ObjectCacheEntry& entry,
+  void touch_lru(const string& name, ObjectCacheEntry& entry,
 		 std::list<string>::iterator& lru_iter);
-  void remove_lru(string& name, std::list<string>::iterator& lru_iter);
+  void remove_lru(const string& name, std::list<string>::iterator& lru_iter);
   void invalidate_lru(ObjectCacheEntry& entry);
 
   void do_invalidate_all();
 public:
   ObjectCache() : lru_size(0), lru_counter(0), lru_window(0), lock("ObjectCache"), cct(NULL), enabled(false) { }
-  int get(std::string& name, ObjectCacheInfo& bl, uint32_t mask, rgw_cache_entry_info *cache_info);
-  void put(std::string& name, ObjectCacheInfo& bl, rgw_cache_entry_info *cache_info);
-  void remove(std::string& name);
+  int get(const std::string& name, ObjectCacheInfo& bl, uint32_t mask, rgw_cache_entry_info *cache_info);
+  std::optional<ObjectCacheInfo> get(const std::string& name) {
+    std::optional<ObjectCacheInfo> info{std::in_place};
+    auto r = get(name, *info, 0, nullptr);
+    return r < 0 ? std::nullopt : info;
+  }
+
+  template<typename F>
+  void for_each(const F& f) {
+    RWLock::RLocker l(lock);
+    if (enabled) {
+      auto now  = ceph::coarse_mono_clock::now();
+      for (const auto& [name, entry] : cache_map) {
+        if (expiry.count() && (now - entry.info.time_added) < expiry) {
+          f(name, entry);
+        }
+      }
+    }
+  }
+
+  void put(const std::string& name, ObjectCacheInfo& bl, rgw_cache_entry_info *cache_info);
+  bool remove(const std::string& name);
   void set_ctx(CephContext *_cct) {
     cct = _cct;
     lru_window = cct->_conf->rgw_cache_lru_size / 2;
@@ -254,6 +273,11 @@ class RGWCache  : public T
   bool chain_cache_entry(std::initializer_list<rgw_cache_entry_info *> cache_info_entries, RGWChainedCache::Entry *chained_entry) override {
     return cache.chain_cache_entry(cache_info_entries, chained_entry);
   }
+  void call_list(const std::optional<std::string>& filter,
+		 Formatter* format) override;
+  bool call_inspect(const std::string& target, Formatter* format) override;
+  bool call_erase(const std::string& target) override;
+  void call_zap() override;
 };
 
 template <class T>
@@ -589,4 +613,42 @@ int RGWCache<T>::watch_cb(uint64_t notify_id,
   return 0;
 }
 
+template<typename T>
+void RGWCache<T>::call_list(const std::optional<std::string>& filter,
+			    Formatter* f)
+{
+  cache.for_each(
+    [this, &filter, f] (const string& name, const ObjectCacheEntry& entry) {
+      if (!filter || name.find(*filter) != name.npos) {
+	T::cache_list_dump_helper(f, name, entry.info.meta.mtime,
+				  entry.info.meta.size);
+      }
+    });
+}
+
+template<typename T>
+bool RGWCache<T>::call_inspect(const std::string& target, Formatter* f)
+{
+  if (const auto entry = cache.get(target)) {
+    f->open_object_section("cache_entry");
+    f->dump_string("name", target.c_str());
+    entry->dump(f);
+    f->close_section();
+    return true;
+  } else {
+    return false;
+  }
+}
+
+template<typename T>
+bool RGWCache<T>::call_erase(const std::string& target)
+{
+  return cache.remove(target);
+}
+
+template<typename T>
+void RGWCache<T>::call_zap()
+{
+  cache.invalidate_all();
+}
 #endif

src/rgw/rgw_rados.cc

@@ -1,4 +1,3 @@
-
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
 // vim: ts=8 sw=2 smarttab
 
@@ -3663,6 +3662,15 @@ bool RGWIndexCompletionManager::handle_completion(completion_t cb, complete_op_d
 
 void RGWRados::finalize()
 {
+  auto admin_socket = cct->get_admin_socket();
+  for (auto cmd : admin_commands) {
+    int r = admin_socket->unregister_command(cmd[0]);
+    if (r < 0) {
+      lderr(cct) << "ERROR: fail to unregister admin socket command (r=" << r
+                 << ")" << dendl;
+    }
+  }
+
   if (run_sync_thread) {
     Mutex::Locker l(meta_sync_thread_lock);
     meta_sync_processor_thread->stop();
@@ -3768,6 +3776,17 @@ void RGWRados::finalize()
 int RGWRados::init_rados()
 {
   int ret = 0;
+  auto admin_socket = cct->get_admin_socket();
+  for (auto cmd : admin_commands) {
+    int r = admin_socket->register_command(cmd[0], cmd[1], this,
+                                           cmd[2]);
+    if (r < 0) {
+      lderr(cct) << "ERROR: fail to register admin socket command (r=" << r
+                 << ")" << dendl;
+      return r;
+    }
+  }
+
   auto handles = std::vector<librados::Rados>{cct->_conf->rgw_num_rados_handles};
 
   for (auto& r : handles) {
@@ -13991,3 +14010,70 @@ int rgw_compression_info_from_attrset(map<string, bufferlist>& attrs, bool& need
   }
 }
 
+bool RGWRados::call(std::string command, cmdmap_t& cmdmap, std::string format,
+                    bufferlist& out)
+{
+  if (command == "cache list"sv) {
+    std::optional<std::string> filter;
+    if (auto i = cmdmap.find("filter"); i != cmdmap.cend()) {
+      filter = boost::get<std::string>(i->second);
+    }
+    std::unique_ptr<Formatter> f(ceph::Formatter::create(format, "table"));
+    if (f) {
+      f->open_array_section("cache_entries");
+      call_list(filter, f.get());
+      f->close_section();
+      f->flush(out);
+      return true;
+    } else {
+      out.append("Unable to create Formatter.\n");
+      return false;
+    }
+  } else if (command == "cache inspect"sv) {
+    std::unique_ptr<Formatter> f(ceph::Formatter::create(format, "json-pretty"));
+    if (f) {
+      const auto& target = boost::get<std::string>(cmdmap["target"]);
+      if (call_inspect(target, f.get())) {
+        f->flush(out);
+        return true;
+      } else {
+        out.append("Unable to find entry "s + target + ".\n");
+        return false;
+      }
+    } else {
+      out.append("Unable to create Formatter.\n");
+      return false;
+    }
+  } else if (command == "cache erase"sv) {
+    const auto& target = boost::get<std::string>(cmdmap["target"]);
+    if (call_erase(target)) {
+      return true;
+    } else {
+      out.append("Unable to find entry "s + target + ".\n");
+      return false;
+    }
+  } else if (command == "cache zap"sv) {
+    call_zap();
+    return true;
+  }
+  return false;
+}
+
+void RGWRados::call_list(const std::optional<std::string>&,
+                         ceph::Formatter*)
+{
+  return;
+}
+
+bool RGWRados::call_inspect(const std::string&, Formatter*)
+{
+  return false;
+}
+
+bool RGWRados::call_erase(const std::string&) {
+  return false;
+}
+
+void RGWRados::call_zap() {
+  return;
+}

src/rgw/rgw_rados.h

@@ -8,6 +8,7 @@
 
 #include "include/rados/librados.hpp"
 #include "include/Context.h"
+#include "common/admin_socket.h"
 #include "common/RefCountedObj.h"
 #include "common/RWLock.h"
 #include "common/ceph_time.h"
@@ -2225,7 +2226,7 @@ struct tombstone_entry {
 
 class RGWIndexCompletionManager;
 
-class RGWRados
+class RGWRados : public AdminSocketHook
 {
   friend class RGWGC;
   friend class RGWMetaNotifier;
@@ -2241,6 +2242,21 @@ class RGWRados
   friend class BucketIndexLockGuard;
   friend class RGWCompleteMultipart;
 
+  static constexpr const char* admin_commands[4][3] = {
+    { "cache list",
+      "cache list name=filter,type=CephString,req=false",
+      "cache list [filter_str]: list object cache, possibly matching substrings" },
+    { "cache inspect",
+      "cache inspect name=target,type=CephString,req=true",
+      "cache inspect target: print cache element" },
+    { "cache erase",
+      "cache erase name=target,type=CephString,req=true",
+      "cache erase target: erase element from cache" },
+    { "cache zap",
+      "cache zap",
+      "cache zap: erase all elements from cache" }
+  };
+
   /** Open the pool used as root for this gateway */
   int open_root_pool_ctx();
   int open_gc_pool_ctx();
@@ -3432,6 +3448,38 @@ class RGWRados
 		       boost::optional<obj_version> refresh_version);
 public:
 
+  bool call(std::string command, cmdmap_t& cmdmap, std::string format,
+	    bufferlist& out) override final;
+
+protected:
+  void cache_list_dump_helper(Formatter* f,
+			      const std::string& name,
+			      const ceph::real_time mtime,
+			      const std::uint64_t size) {
+    f->dump_string("name", name);
+    f->dump_stream("mtime") << mtime;
+    f->dump_unsigned("size", size);
+  }
+
+  // `call_list` must iterate over all cache entries and call
+  // `cache_list_dump_helper` with the supplied Formatter on any that
+  // include `filter` as a substring.
+  //
+  virtual void call_list(const std::optional<std::string>& filter,
+			 Formatter* format);
+  // `call_inspect` must look up the requested target and, if found,
+  // dump it to the supplied Formatter and return true. If not found,
+  // it must return false.
+  //
+  virtual bool call_inspect(const std::string& target, Formatter* format);
+
+  // `call_erase` must erase the requested target and return true. If
+  // the requested target does not exist, it should return false.
+  virtual bool call_erase(const std::string& target);
+
+  // `call_zap` must erase the cache.
+  virtual void call_zap();
+public:
 
   int get_bucket_info(RGWObjectCtx& obj_ctx,
 		      const string& tenant_name, const string& bucket_name,