mon/OSDMonitor: implement mon_max_pool_pg_num_per_osd limit

Prevent pg_num from being set to too many pgs relative to the
number of OSDs.  Take the pool size (replication factor or ec k+m) into
consideration.

Default to the recommended value (~100 pgs/osd).  If the user really wants
more, they can change the configurable limit.

Implement safety check for both pool creation and pg_num changes.

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

src/common/options.cc

@@ -1477,6 +1477,11 @@ std::vector<Option> get_global_options() {
     .set_default(65536)
     .set_description(""),
 
+    Option("mon_max_pool_pg_num_per_osd", Option::TYPE_INT, Option::LEVEL_ADVANCED)
+    .set_default(100)
+    .set_description("Max number of PGs per OSD we allow when creating pools or adjusting pg_num.")
+    .set_long_description("This value reflects *instances* of the PG, so that the replication factor or erasure code K and M values are fully taken into consideration."),
+
     Option("mon_pool_quota_warn_threshold", Option::TYPE_INT, Option::LEVEL_ADVANCED)
     .set_default(0)
     .set_description(""),

src/mon/OSDMonitor.cc

@@ -5797,6 +5797,22 @@ int OSDMonitor::get_crush_rule(const string &rule_name,
   return 0;
 }
 
+int OSDMonitor::check_pg_num(int pg_num, int size, ostream *ss)
+{
+  int64_t max_pgs_per_osd =
+    g_conf->get_val<int64_t>("mon_max_pool_pg_num_per_osd");
+  int64_t max_pgs = max_pgs_per_osd * osdmap.get_num_osds();
+  int64_t max_pg_num = MAX(1, max_pgs / size);
+  if (pg_num > max_pg_num) {
+    *ss << "pg_num " << pg_num << " > max " << max_pg_num << ", as determined"
+	<< " by mon_max_pool_pg_num_per_osd " << max_pgs_per_osd
+	<< " * num_osds " << osdmap.get_num_osds()
+	<< " / pool size " << size;
+    return -ERANGE;
+  }
+  return 0;
+}
+
 /**
  * @param name The name of the new pool
  * @param auid The auid of the pool owner. Can be -1
@@ -5876,6 +5892,11 @@ int OSDMonitor::prepare_new_pool(string& name, uint64_t auid,
     dout(10) << " prepare_pool_size returns " << r << dendl;
     return r;
   }
+  r = check_pg_num(pg_num, size, ss);
+  if (r) {
+    dout(10) << " prepare_pool_size returns " << r << dendl;
+    return r;
+  }
 
   if (!osdmap.crush->check_crush_rule(crush_rule, pool_type, size, *ss)) {
     return -EINVAL;
@@ -6121,6 +6142,10 @@ int OSDMonitor::prepare_command_pool_set(map<string,cmd_vartype> &cmdmap,
          << " (you may adjust 'mon max pool pg num' for higher values)";
       return -ERANGE;
     }
+    int r = check_pg_num(n, p.get_size(), &ss);
+    if (r) {
+      return r;
+    }
     string force;
     cmd_getval(g_ceph_context,cmdmap, "force", force);
     if (p.cache_mode != pg_pool_t::CACHEMODE_NONE &&

src/mon/OSDMonitor.h

@@ -346,6 +346,7 @@ class OSDMonitor : public PaxosService {
 				const string &erasure_code_profile,
 				unsigned *stripe_width,
 				ostream *ss);
+  int check_pg_num(int pg_num, int size, ostream* ss);
   int prepare_new_pool(string& name, uint64_t auid,
 		       int crush_rule,
 		       const string &crush_rule_name,