osd: try remove more stale osdmaps

in a large cluster, there are better chances that the OSD fails to trim
the cached osdmap in a timely manner. and sometimes, it is just unable
to keep up with the incoming osdmap, so the osdmap cache can keep
building up to over 250GB in size. in this change

* publish_superblock() before trimming the osdmaps, so other osdmap
  consumers of OSDService.superblock won't access the osdmaps being
  removed.
* trim all stale osdmaps in batch of conf->osd_target_transaction_size
  without enforcing an upper bound of
  max(conf->osd_target_transaction_size, m->get_last() - m->get_first())
  we had before.
* always use dedicated transaction(s) for trimming osdmaps. so even in
  the normal case where we are able to trim all stale osdmaps in a
  single batch, a separated transaction is used. we can piggy back
  the commits for removing maps, but we keep it this way for simplicity.
* use std::min() instead MIN() for type safety

Fixes: ceph#13990
Signed-off-by: Kefu Chai <kchai@redhat.com>

src/osd/OSD.cc

@@ -6134,6 +6134,41 @@ void OSD::osdmap_subscribe(version_t epoch, bool force_request)
   }
 }
 
+void OSD::trim_maps(epoch_t oldest)
+{
+  if (!superblock.oldest_map)
+    return;
+
+  epoch_t min = std::min(oldest, service.map_cache.cached_key_lower_bound());
+  if (superblock.oldest_map >= min)
+    return;
+
+  int num = 0;
+  ObjectStore::Transaction *t = NULL;
+  for (epoch_t e = superblock.oldest_map; e < min; ++e) {
+    dout(20) << " removing old osdmap epoch " << e << dendl;
+    if (!t) {
+      t = new ObjectStore::Transaction;
+    }
+    t->remove(META_COLL, get_osdmap_pobject_name(e));
+    t->remove(META_COLL, get_inc_osdmap_pobject_name(e));
+    superblock.oldest_map = e + 1;
+    num++;
+    if (num >= cct->_conf->osd_target_transaction_size) {
+      service.publish_superblock(superblock);
+      // handle_osd_map() will write_superblock(), so we can save it for now.
+      store->queue_transaction_and_cleanup(NULL, t);
+      t = NULL;
+    }
+  }
+  if (t) {
+    service.publish_superblock(superblock);
+    store->queue_transaction_and_cleanup(NULL, t);
+  }
+  // we should not remove the cached maps
+  assert(min <= service.map_cache.cached_key_lower_bound());
+}
+
 void OSD::handle_osd_map(MOSDMap *m)
 {
   assert(osd_lock.is_locked());
@@ -6300,22 +6335,7 @@ void OSD::handle_osd_map(MOSDMap *m)
     return;
   }
 
-  if (superblock.oldest_map) {
-    int num = 0;
-    epoch_t min(
-      MIN(m->oldest_map,
-	  service.map_cache.cached_key_lower_bound()));
-    for (epoch_t e = superblock.oldest_map; e < min; ++e) {
-      dout(20) << " removing old osdmap epoch " << e << dendl;
-      t.remove(META_COLL, get_osdmap_pobject_name(e));
-      t.remove(META_COLL, get_inc_osdmap_pobject_name(e));
-      superblock.oldest_map = e+1;
-      num++;
-      if (num >= cct->_conf->osd_target_transaction_size &&
-	  (uint64_t)num > (last - first))  // make sure we at least keep pace with incoming maps
-	break;
-    }
-  }
+  trim_maps(m->oldest_map);
 
   if (!superblock.oldest_map || skip_maps)
     superblock.oldest_map = first;

src/osd/OSD.h

@@ -1658,6 +1658,7 @@ class OSD : public Dispatcher,
 
   void wait_for_new_map(OpRequestRef op);
   void handle_osd_map(class MOSDMap *m);
+  void trim_maps(epoch_t oldest);
   void note_down_osd(int osd);
   void note_up_osd(int osd);