mgr/DaemonServer: set messenger policy; set throttles

Put clients, osds, mons, and mdss in different buckets.

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 8c22e3837824dc8af32acee3028a5d39d5d4a077..a0b77264aa3f8d6d1da2435896a6e7e5be538857 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -1628,11 +1628,21 @@ OPTION(rgw_period_push_interval, OPT_DOUBLE, 2) // seconds to wait before retryi
 OPTION(rgw_period_push_interval_max, OPT_DOUBLE, 30) // maximum interval after exponential backoff
 
 OPTION(rgw_swift_versioning_enabled, OPT_BOOL, false) // whether swift object versioning feature is enabled
+
 OPTION(mgr_module_path, OPT_STR, CEPH_PKGLIBDIR "/mgr") // where to load python modules from
 OPTION(mgr_modules, OPT_STR, "rest")  // Which modules to load
 OPTION(mgr_data, OPT_STR, "/var/lib/ceph/mgr/$cluster-$id") // where to find keyring etc
 OPTION(mgr_beacon_period, OPT_INT, 5)  // How frequently to send beacon
 OPTION(mgr_stats_period, OPT_INT, 5) // How frequently to send stats
+OPTION(mgr_client_bytes, OPT_U64, 128*1048576) // bytes from clients
+OPTION(mgr_client_messages, OPT_U64, 512)      // messages from clients
+OPTION(mgr_osd_bytes, OPT_U64, 512*1048576)   // bytes from osds
+OPTION(mgr_osd_messages, OPT_U64, 8192)       // messages from osds
+OPTION(mgr_mds_bytes, OPT_U64, 128*1048576)   // bytes from mdss
+OPTION(mgr_mds_messages, OPT_U64, 128)        // messages from mdss
+OPTION(mgr_mon_bytes, OPT_U64, 128*1048576)   // bytes from mons
+OPTION(mgr_mon_messages, OPT_U64, 128)        // messages from mons
+
 OPTION(mon_mgr_digest_period, OPT_INT, 5)  // How frequently to send digests
 OPTION(mon_mgr_beacon_grace, OPT_INT, 30)  // How long to wait to failover
 

diff --git a/src/mgr/DaemonServer.cc b/src/mgr/DaemonServer.cc
index 88476cac79fa2209362f5dee9bf8fb4956b0db30..65270fe401527d2adee6fcc20f51d366266e8f62 100644 (file)
--- a/src/mgr/DaemonServer.cc
+++ b/src/mgr/DaemonServer.cc
@@ -30,7 +30,25 @@ DaemonServer::DaemonServer(MonClient *monc_,
   DaemonStateIndex &daemon_state_,
   ClusterState &cluster_state_,
   PyModules &py_modules_)
-    : Dispatcher(g_ceph_context), msgr(nullptr), monc(monc_),
+    : Dispatcher(g_ceph_context),
+      client_byte_throttler(new Throttle(g_ceph_context, "mgr_client_bytes",
+                                        g_conf->mgr_client_bytes)),
+      client_msg_throttler(new Throttle(g_ceph_context, "mgr_client_messages",
+                                       g_conf->mgr_client_messages)),
+      osd_byte_throttler(new Throttle(g_ceph_context, "mgr_osd_bytes",
+                                     g_conf->mgr_osd_bytes)),
+      osd_msg_throttler(new Throttle(g_ceph_context, "mgr_osd_messsages",
+                                    g_conf->mgr_osd_messages)),
+      mds_byte_throttler(new Throttle(g_ceph_context, "mgr_mds_bytes",
+                                     g_conf->mgr_mds_bytes)),
+      mds_msg_throttler(new Throttle(g_ceph_context, "mgr_mds_messsages",
+                                    g_conf->mgr_mds_messages)),
+      mon_byte_throttler(new Throttle(g_ceph_context, "mgr_mon_bytes",
+                                     g_conf->mgr_mon_bytes)),
+      mon_msg_throttler(new Throttle(g_ceph_context, "mgr_mon_messsages",
+                                    g_conf->mgr_mon_messages)),
+      msgr(nullptr),
+      monc(monc_),
       daemon_state(daemon_state_),
       cluster_state(cluster_state_),
       py_modules(py_modules_),
@@ -48,9 +66,30 @@ DaemonServer::~DaemonServer() {
 int DaemonServer::init(uint64_t gid, entity_addr_t client_addr)
 {
   // Initialize Messenger
-  std::string public_msgr_type = g_conf->ms_public_type.empty() ? g_conf->get_val<std::string>("ms_type") : g_conf->ms_public_type;
+  std::string public_msgr_type = g_conf->ms_public_type.empty() ?
+    g_conf->get_val<std::string>("ms_type") : g_conf->ms_public_type;
   msgr = Messenger::create(g_ceph_context, public_msgr_type,
-                          entity_name_t::MGR(gid), "server", getpid(), 0);
+                          entity_name_t::MGR(gid),
+                          "mgr",
+                          getpid(), 0);
+  msgr->set_default_policy(Messenger::Policy::stateless_server(0));
+
+  // throttle clients
+  msgr->set_policy_throttlers(entity_name_t::TYPE_CLIENT,
+                             client_byte_throttler.get(),
+                             client_msg_throttler.get());
+
+  // servers
+  msgr->set_policy_throttlers(entity_name_t::TYPE_OSD,
+                             osd_byte_throttler.get(),
+                             osd_msg_throttler.get());
+  msgr->set_policy_throttlers(entity_name_t::TYPE_MDS,
+                             mds_byte_throttler.get(),
+                             mds_msg_throttler.get());
+  msgr->set_policy_throttlers(entity_name_t::TYPE_MON,
+                             mon_byte_throttler.get(),
+                             mon_msg_throttler.get());
+
   int r = msgr->bind(g_conf->public_addr);
   if (r < 0) {
     derr << "unable to bind mgr to " << g_conf->public_addr << dendl;

diff --git a/src/mgr/DaemonServer.h b/src/mgr/DaemonServer.h
index 91ce05776dafa05bd740e900bfd5751eb5ab6cf0..e38f3379ac26aaf2aa9cc36dd4ae1ca261b91812 100644 (file)
--- a/src/mgr/DaemonServer.h
+++ b/src/mgr/DaemonServer.h
@@ -40,6 +40,15 @@ class MCommand;
 class DaemonServer : public Dispatcher
 {
 protected:
+  boost::scoped_ptr<Throttle> client_byte_throttler;
+  boost::scoped_ptr<Throttle> client_msg_throttler;
+  boost::scoped_ptr<Throttle> osd_byte_throttler;
+  boost::scoped_ptr<Throttle> osd_msg_throttler;
+  boost::scoped_ptr<Throttle> mds_byte_throttler;
+  boost::scoped_ptr<Throttle> mds_msg_throttler;
+  boost::scoped_ptr<Throttle> mon_byte_throttler;
+  boost::scoped_ptr<Throttle> mon_msg_throttler;
+
   Messenger *msgr;
   MonClient *monc;
   DaemonStateIndex &daemon_state;