osd: move PG collection creation into static PG method

This way we properly create the pgmeta object and set the infokey attr on
it.  Break it into two steps, one part before split, one after.

Disable the collection_empty() check in split so that it doesn't trip over
this object.

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

src/os/FileStore.cc

@@ -5081,8 +5081,6 @@ int FileStore::_split_collection(coll_t cid,
     int dstcmp = _check_replay_guard(dest, spos);
     if (dstcmp < 0)
       return 0;
-    if (dstcmp > 0 && !collection_empty(dest))
-      return -ENOTEMPTY;
 
     int srccmp = _check_replay_guard(cid, spos);
     if (srccmp < 0)

src/osd/OSD.cc

@@ -3071,19 +3071,8 @@ void OSD::handle_pg_peering_evt(
     switch (result) {
     case RES_NONE: {
       const pg_pool_t* pp = osdmap->get_pg_pool(pgid.pool());
-      coll_t cid(pgid);
+      PG::_create(*rctx.transaction, pgid);
-
+      PG::_init(*rctx.transaction, pgid, pp);
-      // ok, create the pg locally using provided Info and History
-      rctx.transaction->create_collection(cid);
-
-      // Give a hint to the PG collection
-      bufferlist hint;
-      uint32_t pg_num = pp->get_pg_num();
-      uint64_t expected_num_objects_pg = pp->expected_num_objects / pg_num;
-      ::encode(pg_num, hint);
-      ::encode(expected_num_objects_pg, hint);
-      uint32_t hint_type = ObjectStore::Transaction::COLL_HINT_EXPECTED_NUM_OBJECTS;
-      rctx.transaction->collection_hint(cid, hint_type, hint);
 
       PG *pg = _create_lock_pg(
 	get_map(epoch),
@@ -6807,6 +6796,7 @@ void OSD::split_pgs(
       *i,
       split_bits,
       i->ps(),
+      &child->pool.info,
       rctx->transaction);
     parent->split_into(
       i->pgid,
@@ -6966,19 +6956,10 @@ void OSD::handle_pg_create(OpRequestRef op)
     PG *pg = NULL;
     if (can_create_pg(pgid)) {
       const pg_pool_t* pp = osdmap->get_pg_pool(pgid.pool());
+      PG::_create(*rctx.transaction, pgid);
+      PG::_init(*rctx.transaction, pgid, pp);
+
       pg_interval_map_t pi;
-      coll_t cid(pgid);
-      rctx.transaction->create_collection(cid);
-
-      // Give a hint to the PG collection
-      bufferlist hint;
-      uint32_t pg_num = pp->get_pg_num();
-      uint64_t expected_num_objects_pg = pp->expected_num_objects / pg_num;
-      ::encode(pg_num, hint);
-      ::encode(expected_num_objects_pg, hint);
-      uint32_t hint_type = ObjectStore::Transaction::COLL_HINT_EXPECTED_NUM_OBJECTS;
-      rctx.transaction->collection_hint(cid, hint_type, hint);
-
       pg = _create_lock_pg(
 	osdmap, pgid, true, false, false,
 	0, creating_pgs[pgid].acting, whoami,

src/osd/PG.cc

@@ -2639,6 +2639,28 @@ int PG::_write_info(ObjectStore::Transaction& t, epoch_t epoch,
   return 0;
 }
 
+void PG::_create(ObjectStore::Transaction& t, spg_t pgid)
+{
+  coll_t coll(pgid);
+  t.create_collection(coll);
+}
+
+void PG::_init(ObjectStore::Transaction& t, spg_t pgid, const pg_pool_t *pool)
+{
+  coll_t coll(pgid);
+
+  if (pool) {
+    // Give a hint to the PG collection
+    bufferlist hint;
+    uint32_t pg_num = pool->get_pg_num();
+    uint64_t expected_num_objects_pg = pool->expected_num_objects / pg_num;
+    ::encode(pg_num, hint);
+    ::encode(expected_num_objects_pg, hint);
+    uint32_t hint_type = ObjectStore::Transaction::COLL_HINT_EXPECTED_NUM_OBJECTS;
+    t.collection_hint(coll, hint_type, hint);
+  }
+}
+
 void PG::write_info(ObjectStore::Transaction& t)
 {
   info.stats.stats.add(unstable_stats);

src/osd/PG.h

@@ -1202,6 +1202,7 @@ public:
     spg_t child,
     int split_bits,
     int seed,
+    const pg_pool_t *pool,
     ObjectStore::Transaction *t) = 0;
   virtual bool _report_snap_collection_errors(
     const hobject_t &hoid,
@@ -2087,6 +2088,10 @@ public:
   // pg on-disk state
   void do_pending_flush();
 
+  static void _create(ObjectStore::Transaction& t, spg_t pgid);
+  static void _init(ObjectStore::Transaction& t,
+		    spg_t pgid, const pg_pool_t *pool);
+
 private:
   void write_info(ObjectStore::Transaction& t);
 

src/osd/ReplicatedPG.h

@@ -1326,14 +1326,16 @@ public:
     spg_t child,
     int split_bits,
     int seed,
+    const pg_pool_t *pool,
     ObjectStore::Transaction *t) {
     coll_t target = coll_t(child);
-    t->create_collection(target);
+    PG::_create(*t, child);
     t->split_collection(
       coll,
       split_bits,
       seed,
       target);
+    PG::_init(*t, child, pool);
     pgbackend->split_colls(child, split_bits, seed, t);
   }
 private:

src/tools/ceph_objectstore_tool.cc

@@ -1583,8 +1583,12 @@ int do_import(ObjectStore *store, OSDSuperblock& sb)
   bufferlist one;
   one.append('1');
   ObjectStore::Transaction *t = new ObjectStore::Transaction;
-  t->create_collection(coll);
+  PG::_create(*t, pgid);
+  PG::_init(*t, pgid, NULL);
+
+  // mark this coll for removal until we're done
   t->collection_setattr(coll, "remove", one);
+
   store->apply_transaction(*t);
   delete t;