From: Jos Collin <jcollin@redhat.com>
Date: Wed, 4 Nov 2020 13:21:44 +0000 (+0530)
Subject: msg,mon,common: log when DispatchQueue throttle limit is reached
X-Git-Tag: v17.1.0~3079^2~1
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=198e79a02471331b09750395f257e88879e053e4;p=ceph.git

msg,mon,common: log when DispatchQueue throttle limit is reached

Fixes: https://tracker.ceph.com/issues/46226
Signed-off-by: Jos Collin <jcollin@redhat.com>
---

diff --git a/src/common/options.cc b/src/common/options.cc
index 8e15babf9e3c..7f63843ae50d 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -1019,6 +1019,11 @@ std::vector<Option> get_global_options() {
     .set_default(100_M)
     .set_description("Limit messages that are read off the network but still being processed"),
 
+    Option("ms_dispatch_throttle_log_interval", Option::TYPE_SECS, Option::LEVEL_ADVANCED)
+    .set_default(30)
+    .set_min(0)
+    .set_description("Interval in seconds for high verbosity debug log message when the dispatch throttle limit are hit"),
+
     Option("ms_bind_ipv4", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
     .set_default(true)
     .set_description("Bind servers to IPv4 address(es)")
diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
index a1c67ad44990..1e9087aff82e 100644
--- a/src/mon/MonClient.cc
+++ b/src/mon/MonClient.cc
@@ -508,6 +508,7 @@ int MonClient::init()
 
   initialized = true;
 
+  cct->_conf.add_observer(this);
   messenger->set_auth_client(this);
   messenger->add_dispatcher_head(this);
 
@@ -546,6 +547,7 @@ void MonClient::shutdown()
   if (initialized) {
     initialized = false;
   }
+  cct->_conf.remove_observer(this);
   monc_lock.lock();
   timer.shutdown();
   stopping = false;
@@ -854,6 +856,29 @@ bool MonClient::ms_handle_reset(Connection *con)
   }
 }
 
+bool MonClient::ms_handle_throttle(ms_throttle_t ttype) {
+  switch (ttype) {
+  case ms_throttle_t::MESSAGE:
+    break; // TODO
+  case ms_throttle_t::BYTES:
+    break; // TODO
+  case ms_throttle_t::DISPATCH_QUEUE:
+    {
+      //cluster log a warning that Dispatch Queue Throttle Limit hit
+      if (!log_client) {
+        return false; //cannot handle if the daemon didn't setup a log_client for me
+      }
+      LogChannelRef clog = log_client->create_channel(CLOG_CHANNEL_CLUSTER);
+      clog->warn() << "Throttler Limit has been hit. "
+                   << "Some message processing may be significantly delayed.";
+    }
+    break;
+  default:
+    return false;
+  }
+  return true;
+}
+
 bool MonClient::_opened() const
 {
   ceph_assert(ceph_mutex_is_locked(monc_lock));
@@ -1615,6 +1640,24 @@ int MonClient::handle_auth_request(
   return -EACCES;
 }
 
+const char** MonClient::get_tracked_conf_keys() const {
+  static const char* KEYS[] = {
+    "ms_dispatch_throttle_bytes",
+    "ms_dispatch_throttle_log_interval",
+    NULL
+  };
+  return KEYS;
+}
+
+void MonClient::handle_conf_change(const ConfigProxy& conf, const std::set<std::string> &changed) {
+  if (changed.count("ms_dispatch_throttle_bytes") || changed.count("ms_dispatch_throttle_log_interval")) {
+    if (messenger) {
+      messenger->dispatch_throttle_bytes = cct->_conf.get_val<Option::size_t>("ms_dispatch_throttle_bytes");
+      messenger->dispatch_throttle_log_interval = cct->_conf.get_val<std::chrono::seconds>("ms_dispatch_throttle_log_interval");
+    }
+  }
+}
+
 AuthAuthorizer* MonClient::build_authorizer(int service_id) const {
   std::lock_guard l(monc_lock);
   if (auth) {
diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h
index cc7805d0675e..7231d1a65549 100644
--- a/src/mon/MonClient.h
+++ b/src/mon/MonClient.h
@@ -268,8 +268,9 @@ inline boost::system::error_condition make_error_condition(monc_errc e) noexcept
 const boost::system::error_category& monc_category() noexcept;
 
 class MonClient : public Dispatcher,
-		  public AuthClient,
-		  public AuthServer /* for mgr, osd, mds */ {
+                  public AuthClient,
+                  public AuthServer, /* for mgr, osd, mds */
+                  public md_config_obs_t {
   static constexpr auto dout_subsys = ceph_subsys_monc;
 public:
   // Error, Newest, Oldest
@@ -309,6 +310,7 @@ private:
   bool ms_handle_reset(Connection *con) override;
   void ms_handle_remote_reset(Connection *con) override {}
   bool ms_handle_refused(Connection *con) override { return false; }
+  bool ms_handle_throttle(ms_throttle_t ttype) override;
 
   void handle_monmap(MMonMap *m);
   void handle_config(MConfig *m);
@@ -402,6 +404,11 @@ public:
     uint32_t auth_method,
     const ceph::buffer::list& bl,
     ceph::buffer::list *reply) override;
+  // md_config_obs_t (config observer)
+  const char** get_tracked_conf_keys() const override;
+  void handle_conf_change(
+    const ConfigProxy& conf,
+    const std::set<std::string> &changed) override;
 
   void set_entity_name(EntityName name) { entity_name = name; }
   void set_handle_authentication_dispatcher(Dispatcher *d) {
diff --git a/src/msg/DispatchQueue.h b/src/msg/DispatchQueue.h
index de0cb7d1a083..c951df5370df 100644
--- a/src/msg/DispatchQueue.h
+++ b/src/msg/DispatchQueue.h
@@ -212,6 +212,11 @@ class DispatchQueue {
   uint64_t get_id() {
     return next_id++;
   }
+
+  Messenger* get_messenger() const {
+    return msgr;
+  }
+
   void start();
   void entry();
   void wait();
diff --git a/src/msg/Dispatcher.h b/src/msg/Dispatcher.h
index 5e025437b535..36141571202a 100644
--- a/src/msg/Dispatcher.h
+++ b/src/msg/Dispatcher.h
@@ -214,6 +214,16 @@ public:
     return 0;
   }
 
+  /**
+   * handle throttle limit hit and cluster log it.
+   *
+   * return true if handled
+   * return false if not handled
+   */
+  virtual bool ms_handle_throttle(ms_throttle_t ttype) {
+    return false;
+  }
+
   /**
    * @} //Authentication
    */
diff --git a/src/msg/Messenger.cc b/src/msg/Messenger.cc
index 8064a10a0d9b..b701d17f9124 100644
--- a/src/msg/Messenger.cc
+++ b/src/msg/Messenger.cc
@@ -66,6 +66,8 @@ Messenger::Messenger(CephContext *cct_, entity_name_t w)
     auth_registry(cct)
 {
   auth_registry.refresh_config();
+  dispatch_throttle_bytes = cct->_conf.get_val<Option::size_t>("ms_dispatch_throttle_bytes");
+  dispatch_throttle_log_interval = cct->_conf.get_val<std::chrono::seconds>("ms_dispatch_throttle_log_interval");
 }
 
 void Messenger::set_endpoint_addr(const entity_addr_t& a,
diff --git a/src/msg/Messenger.h b/src/msg/Messenger.h
index e87f3196b1c9..7dcaeb75ba48 100644
--- a/src/msg/Messenger.h
+++ b/src/msg/Messenger.h
@@ -112,6 +112,8 @@ protected:
 public:
   AuthClient *auth_client = 0;
   AuthServer *auth_server = 0;
+  uint64_t dispatch_throttle_bytes;
+  std::chrono::seconds dispatch_throttle_log_interval;
 
 #ifdef UNIT_TESTS_BUILT
   Interceptor *interceptor = nullptr;
@@ -813,6 +815,18 @@ public:
   void set_require_authorizer(bool b) {
     require_authorizer = b;
   }
+  /**
+   * Notify each Dispatcher that the Throttle Limit has been hit. Call
+   * this function whenever the connections are getting throttled.
+   *
+   * @param ttype Throttle type
+   */
+  void ms_deliver_throttle(ms_throttle_t ttype) {
+    for (const auto &dispatcher : dispatchers) {
+      if (dispatcher->ms_handle_throttle(ttype))
+        return;
+    }
+  }
 
   /**
    * @} // Dispatcher Interfacing
diff --git a/src/msg/async/Protocol.h b/src/msg/async/Protocol.h
index 10436307ebf8..d9f3db7a778c 100644
--- a/src/msg/async/Protocol.h
+++ b/src/msg/async/Protocol.h
@@ -106,6 +106,7 @@ protected:
   AsyncConnection *connection;
   AsyncMessenger *messenger;
   CephContext *cct;
+  ceph::mono_time throttle_prev = ceph::mono_clock::zero();
 public:
   std::shared_ptr<AuthConnectionMeta> auth_meta;
 
diff --git a/src/msg/async/ProtocolV1.cc b/src/msg/async/ProtocolV1.cc
index 43363371bc35..3f2100968dd4 100644
--- a/src/msg/async/ProtocolV1.cc
+++ b/src/msg/async/ProtocolV1.cc
@@ -718,6 +718,10 @@ CtPtr ProtocolV1::throttle_dispatch_queue() {
   ldout(cct, 20) << __func__ << dendl;
 
   if (cur_msg_size) {
+    Messenger* msgr = connection->dispatch_queue->get_messenger();
+    //update max if it's changed in the conf. Expecting qa tests would change ms_dispatch_throttle_bytes.
+    connection->dispatch_queue->dispatch_throttler.reset_max(msgr->dispatch_throttle_bytes);
+
     if (!connection->dispatch_queue->dispatch_throttler.get_or_fail(
             cur_msg_size)) {
       ldout(cct, 10)
@@ -726,6 +730,16 @@ CtPtr ProtocolV1::throttle_dispatch_queue() {
           << connection->dispatch_queue->dispatch_throttler.get_current() << "/"
           << connection->dispatch_queue->dispatch_throttler.get_max()
           << " failed, just wait." << dendl;
+      ceph::mono_time throttle_now = ceph::mono_clock::now();
+      auto duration = std::chrono::duration_cast<std::chrono::seconds>(throttle_now - throttle_prev);
+      if (duration >= msgr->dispatch_throttle_log_interval) {
+        ldout(cct, 1) << __func__ << " Throttler Limit has been hit. "
+                      << "Some message processing may be significantly delayed." << dendl;
+        throttle_prev = throttle_now;
+
+        //Cluster logging that throttling is occurring.
+        msgr->ms_deliver_throttle(ms_throttle_t::DISPATCH_QUEUE);
+      }
       // following thread pool deal with th full message queue isn't a
       // short time, so we can wait a ms.
       if (connection->register_time_events.empty()) {
diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index 855006447f79..2de83a042380 100644
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -1572,6 +1572,10 @@ CtPtr ProtocolV2::throttle_dispatch_queue() {
 
   const size_t cur_msg_size = get_current_msg_size();
   if (cur_msg_size) {
+    Messenger* msgr = connection->dispatch_queue->get_messenger();
+    //update max if it's changed in the conf. Expecting qa tests would change ms_dispatch_throttle_bytes.
+    connection->dispatch_queue->dispatch_throttler.reset_max(msgr->dispatch_throttle_bytes);
+
     if (!connection->dispatch_queue->dispatch_throttler.get_or_fail(
             cur_msg_size)) {
       ldout(cct, 10)
@@ -1580,6 +1584,16 @@ CtPtr ProtocolV2::throttle_dispatch_queue() {
           << connection->dispatch_queue->dispatch_throttler.get_current() << "/"
           << connection->dispatch_queue->dispatch_throttler.get_max()
           << " failed, just wait." << dendl;
+      ceph::mono_time throttle_now = ceph::mono_clock::now();
+      auto duration = std::chrono::duration_cast<std::chrono::seconds>(throttle_now - throttle_prev);
+      if (duration >= msgr->dispatch_throttle_log_interval) {
+        ldout(cct, 1) << __func__ << " Throttler Limit has been hit. "
+                      << "Some message processing may be significantly delayed." << dendl;
+        throttle_prev = throttle_now;
+
+        //Cluster logging that throttling is occurring.
+        msgr->ms_deliver_throttle(ms_throttle_t::DISPATCH_QUEUE);
+      }
       // following thread pool deal with th full message queue isn't a
       // short time, so we can wait a ms.
       if (connection->register_time_events.empty()) {
diff --git a/src/msg/msg_types.h b/src/msg/msg_types.h
index 76a1c1ac2a76..acde32f6ecfe 100644
--- a/src/msg/msg_types.h
+++ b/src/msg/msg_types.h
@@ -811,4 +811,10 @@ inline std::ostream& operator<<(std::ostream& out, const ceph_entity_inst &i)
   return out << n;
 }
 
+enum class ms_throttle_t {
+    MESSAGE,
+    BYTES,
+    DISPATCH_QUEUE
+};
+
 #endif