Merge pull request ceph#56534 from athanatos/sjust/wip-64975-crimson-scrub-snapmapper-2

crimson: introduce INTERNAL_PG_LOCAL_NS, use for snapmapper

Reviewed-by: Yingxin Cheng <[email protected]>

Author: athanatos

src/common/hobject.h

@@ -313,6 +313,26 @@ struct hobject_t {
     return nspace;
   }
 
+  /**
+   * PG_LOCAL_NS
+   *
+   * Used exclusively by crimson at this time.
+   *
+   * Namespace for objects maintained by the local pg instantiation updated
+   * independently of the pg log.  librados IO to this namespace should fail.
+   * Listing operations related to pg objects should exclude objects in this
+   * namespace along with temp objects, ec rollback objects, and the pg
+   * meta object. Such operations include:
+   * - scrub
+   * - backfill
+   * - pgls
+   * See crimson/osd/pg_backend PGBackend::list_objects
+   */
+  static constexpr std::string_view INTERNAL_PG_LOCAL_NS = ".internal_pg_local";
+  bool is_internal_pg_local() const {
+    return nspace == INTERNAL_PG_LOCAL_NS;
+  }
+
   bool parse(const std::string& s);
 
   void encode(ceph::buffer::list& bl) const;
@@ -378,10 +398,10 @@ struct formatter<hobject_t> {
     return ctx.out();
   }
 
-  constexpr auto parse(format_parse_context& ctx) { return ctx.begin(); }
+  constexpr auto parse(format_parse_context& ctx) const { return ctx.begin(); }
 
   template <typename FormatContext>
-  auto format(const hobject_t& ho, FormatContext& ctx)
+  auto format(const hobject_t& ho, FormatContext& ctx) const
   {
     if (ho == hobject_t{}) {
       return fmt::format_to(ctx.out(), "MIN");
@@ -482,6 +502,30 @@ struct ghobject_t {
     return hobj.pool >= 0 && hobj.oid.name.empty();
   }
 
+  bool is_internal_pg_local() const {
+    return hobj.is_internal_pg_local();
+  }
+
+  /**
+   * SNAPMAPPER_OID, make_snapmapper, is_snapmapper
+   *
+   * Used exclusively by crimson at this time.
+   * 
+   * Unlike classic, crimson uses a snap mapper object for each pg.
+   * The snapmapper object provides an index for efficient trimming of clones as
+   * snapshots are removed.
+   *
+   * As with the pgmeta object, we pin the hash to the pg hash.
+   */
+  static constexpr std::string_view SNAPMAPPER_OID = "snapmapper";
+  static ghobject_t make_snapmapper(
+    int64_t pool, uint32_t hash, shard_id_t shard) {
+    hobject_t h(object_t(SNAPMAPPER_OID), std::string(),
+		CEPH_NOSNAP, hash, pool,
+		std::string(hobject_t::INTERNAL_PG_LOCAL_NS));
+    return ghobject_t(h, NO_GEN, shard);
+  }
+
   bool match(uint32_t bits, uint32_t match) const {
     return hobj.match_hash(hobj.hash, bits, match);
   }

src/crimson/osd/ops_executer.cc

@@ -1164,10 +1164,6 @@ static PG::interruptible_future<hobject_t> pgls_filter(
   }
 }
 
-static inline bool is_snapmapper_oid(const hobject_t &obj) {
-  return obj.oid.name == SNAPMAPPER_OID;
-}
-
 static PG::interruptible_future<ceph::bufferlist> do_pgnls_common(
   const hobject_t& pg_start,
   const hobject_t& pg_end,
@@ -1188,13 +1184,6 @@ static PG::interruptible_future<ceph::bufferlist> do_pgnls_common(
     [&backend, filter, nspace](auto&& ret)
     -> PG::interruptible_future<std::tuple<std::vector<hobject_t>, hobject_t>> {
       auto& [objects, next] = ret;
-      auto is_snapmapper = [](const hobject_t &obj) {
-	if (is_snapmapper_oid(obj)) {
-	  return false;
-	} else {
-	  return true;
-	}
-      };
       auto in_my_namespace = [&nspace](const hobject_t& obj) {
         using crimson::common::local_conf;
         if (obj.get_namespace() == local_conf()->osd_hit_set_namespace) {
@@ -1222,8 +1211,7 @@ static PG::interruptible_future<ceph::bufferlist> do_pgnls_common(
         }
       };
 
-      auto range = objects | boost::adaptors::filtered(is_snapmapper)
-			   | boost::adaptors::filtered(in_my_namespace)
+      auto range = objects | boost::adaptors::filtered(in_my_namespace)
                            | boost::adaptors::transformed(to_pglsed);
       logger().debug("do_pgnls_common: finishing the 1st stage of pgls");
       return seastar::when_all_succeed(std::begin(range),
@@ -1356,9 +1344,6 @@ static PG::interruptible_future<ceph::bufferlist> do_pgls_common(
         PG::interruptor::map_reduce(std::move(objects),
           [&backend, filter, nspace](const hobject_t& obj)
 	  -> PG::interruptible_future<hobject_t>{
-	    if (is_snapmapper_oid(obj)) {
-	      return seastar::make_ready_future<hobject_t>();
-	    }
             if (obj.get_namespace() == nspace) {
               if (filter) {
                 return pgls_filter(*filter, backend, obj);

src/crimson/osd/osd_operations/client_request.cc

@@ -324,6 +324,9 @@ ClientRequest::do_process(
     return reply_op_error(pg, -ENAMETOOLONG);
   } else if (m->get_hobj().oid.name.empty()) {
     return reply_op_error(pg, -EINVAL);
+  } else if (m->get_hobj().is_internal_pg_local()) {
+    // clients are not allowed to write to hobject_t::INTERNAL_PG_LOCAL_NS
+    return reply_op_error(pg, -EINVAL);
   } else if (pg->get_osdmap()->is_blocklisted(
         get_foreign_connection().get_peer_addr())) {
     DEBUGDPP("{}.{}: {} is blocklisted",

src/crimson/osd/osd_operations/scrub_events.cc

@@ -2,6 +2,7 @@
 // vim: ts=8 sw=2 smarttab
 
 #include "crimson/common/log.h"
+#include "crimson/common/coroutine.h"
 #include "crimson/osd/pg.h"
 #include "crimson/osd/osd_connection_priv.h"
 #include "messages/MOSDRepScrubMap.h"
@@ -88,24 +89,17 @@ ScrubFindRange::ifut<> ScrubFindRange::run(PG &pg)
 {
   LOG_PREFIX(ScrubFindRange::run);
   using crimson::common::local_conf;
-  return interruptor::make_interruptible(
-    pg.shard_services.get_store().list_objects(
-      pg.get_collection_ref(),
-      ghobject_t(begin, ghobject_t::NO_GEN, pg.get_pgid().shard),
-      ghobject_t::get_max(),
-      local_conf().get_val<int64_t>("osd_scrub_chunk_max")
-    )
-  ).then_interruptible([FNAME, this, &pg](auto ret) {
-    auto &[_, next] = ret;
+  auto [_, next] = co_await pg.backend->list_objects(
+    begin,
+    local_conf().get_val<int64_t>("osd_scrub_chunk_max"));
 
-    // We rely on seeing an entire set of snapshots in a single chunk
-    auto end = next.hobj.get_max_object_boundary();
+  // We rely on seeing an entire set of snapshots in a single chunk
+  auto end = next.get_max_object_boundary();
 
-    DEBUGDPP("got next.hobj: {}, returning begin, end: {}, {}",
-	     pg, next.hobj, begin, end);
-    pg.scrubber.machine.process_event(
-      scrub::ScrubContext::request_range_complete_t{begin, end});
-  });
+  DEBUGDPP("got next: {}, returning begin, end: {}, {}",
+	   pg, next, begin, end);
+  pg.scrubber.machine.process_event(
+    scrub::ScrubContext::request_range_complete_t{begin, end});
 }
 
 template class ScrubAsyncOpT<ScrubFindRange>;
@@ -151,50 +145,36 @@ ScrubScan::ifut<> ScrubScan::run(PG &pg)
   ret.valid_through = pg.get_info().last_update;
 
   DEBUGDPP("begin: {}, end: {}", pg, begin, end);
-  return interruptor::make_interruptible(
-    pg.shard_services.get_store().list_objects(
-      pg.get_collection_ref(),
-      ghobject_t(begin, ghobject_t::NO_GEN, pg.get_pgid().shard),
-      ghobject_t(end, ghobject_t::NO_GEN, pg.get_pgid().shard),
-      std::numeric_limits<uint64_t>::max())
-  ).then_interruptible([FNAME, this, &pg](auto &&result) {
-    DEBUGDPP("listed {} objects", pg, std::get<0>(result).size());
-    return seastar::do_with(
-      std::move(std::get<0>(result)),
-      [this, &pg](auto &objects) {
-	return interruptor::do_for_each(
-	  objects,
-	  [this, &pg](auto &obj) {
-	    if (obj.is_pgmeta() || obj.hobj.is_temp()) {
-	      return interruptor::now();
-	    } else {
-	      return scan_object(pg, obj);
-	    }
-	  });
-      });
-  }).then_interruptible([FNAME, this, &pg] {
-    if (local) {
-      DEBUGDPP("complete, submitting local event", pg);
-      pg.scrubber.handle_event(
-	scrub::ScrubContext::scan_range_complete_t(
-	  pg.get_pg_whoami(),
-	  std::move(ret)));
-      return seastar::now();
-    } else {
-      DEBUGDPP("complete, sending response to primary", pg);
-      auto m = crimson::make_message<MOSDRepScrubMap>(
-	spg_t(pg.get_pgid().pgid, pg.get_primary().shard),
-	pg.get_osdmap_epoch(),
-	pg.get_pg_whoami());
-      encode(ret, m->get_data());
-      pg.scrubber.handle_event(
-	scrub::ScrubContext::generate_and_submit_chunk_result_complete_t{});
-      return pg.shard_services.send_to_osd(
+  auto [objects, _] = co_await pg.backend->list_objects(begin, end);
+
+  DEBUGDPP("listed {} objects", pg, objects);
+  for (const auto &object: objects) {
+    co_await scan_object(
+      pg,
+      ghobject_t(object, ghobject_t::NO_GEN, pg.get_pgid().shard));
+  }
+
+  if (local) {
+    DEBUGDPP("complete, submitting local event", pg);
+    pg.scrubber.handle_event(
+      scrub::ScrubContext::scan_range_complete_t(
+	pg.get_pg_whoami(),
+	std::move(ret)));
+  } else {
+    DEBUGDPP("complete, sending response to primary", pg);
+    auto m = crimson::make_message<MOSDRepScrubMap>(
+      spg_t(pg.get_pgid().pgid, pg.get_primary().shard),
+      pg.get_osdmap_epoch(),
+      pg.get_pg_whoami());
+    encode(ret, m->get_data());
+    pg.scrubber.handle_event(
+      scrub::ScrubContext::generate_and_submit_chunk_result_complete_t{});
+    co_await interruptor::make_interruptible(
+      pg.shard_services.send_to_osd(
 	pg.get_primary().osd,
 	std::move(m),
-	pg.get_osdmap_epoch());
-    }
-  });
+	pg.get_osdmap_epoch()));
+  }
 }
 
 ScrubScan::ifut<> ScrubScan::scan_object(

src/crimson/osd/pg.cc

@@ -136,7 +136,7 @@ PG::PG(
     osdriver(
       &shard_services.get_store(),
       coll_ref,
-      make_snapmapper_oid()),
+      pgid.make_snapmapper_oid()),
     snap_mapper(
       this->shard_services.get_cct(),
       &osdriver,
@@ -618,10 +618,10 @@ seastar::future<> PG::init(
     new_acting_primary, history, pi, t);
   assert(coll_ref);
   return shard_services.get_store().exists(
-    get_collection_ref(), make_snapmapper_oid()
+    get_collection_ref(), pgid.make_snapmapper_oid()
   ).safe_then([&t, this](bool existed) {
       if (!existed) {
-        t.touch(coll_ref->get_cid(), make_snapmapper_oid());
+        t.touch(coll_ref->get_cid(), pgid.make_snapmapper_oid());
       }
     },
     ::crimson::ct_error::assert_all{"unexpected eio"}

src/crimson/osd/pg.h

@@ -41,8 +41,6 @@
 #include "crimson/osd/object_context_loader.h"
 #include "crimson/osd/scrub/pg_scrubber.h"
 
-#define SNAPMAPPER_OID "snapmapper"
-
 class MQuery;
 class OSDMap;
 class PGBackend;
@@ -649,16 +647,6 @@ class PG : public boost::intrusive_ref_counter<
 private:
   OSDriver osdriver;
   SnapMapper snap_mapper;
-  ghobject_t make_snapmapper_oid() const {
-    return ghobject_t(hobject_t(
-      sobject_t(
-       object_t(SNAPMAPPER_OID),
-       0),
-      std::string(),
-      pgid.ps(),
-      pgid.pool(),
-      std::string()));
-  }
 public:
   // PeeringListener
   void publish_stats_to_osd() final;

src/crimson/osd/pg_backend.cc

@@ -17,6 +17,7 @@
 #include "common/Checksummer.h"
 #include "common/Clock.h"
 
+#include "crimson/common/coroutine.h"
 #include "crimson/common/exception.h"
 #include "crimson/common/tmap_helpers.h"
 #include "crimson/os/futurized_collection.h"
@@ -1080,33 +1081,33 @@ PGBackend::remove(ObjectState& os, ceph::os::Transaction& txn,
 }
 
 PGBackend::interruptible_future<std::tuple<std::vector<hobject_t>, hobject_t>>
-PGBackend::list_objects(const hobject_t& start, uint64_t limit) const
+PGBackend::list_objects(
+  const hobject_t& start, const hobject_t &end, uint64_t limit) const
 {
   auto gstart = start.is_min() ? ghobject_t{} : ghobject_t{start, 0, shard};
-  return interruptor::make_interruptible(store->list_objects(coll,
-					 gstart,
-					 ghobject_t::get_max(),
-					 limit))
-    .then_interruptible([](auto ret) {
-      auto& [gobjects, next] = ret;
-      std::vector<hobject_t> objects;
-      boost::copy(gobjects |
-        boost::adaptors::filtered([](const ghobject_t& o) {
-          if (o.is_pgmeta()) {
-            return false;
-          } else if (o.hobj.is_temp()) {
-            return false;
-          } else {
-            return o.is_no_gen();
-          }
-        }) |
-        boost::adaptors::transformed([](const ghobject_t& o) {
-          return o.hobj;
-        }),
-        std::back_inserter(objects));
-      return seastar::make_ready_future<std::tuple<std::vector<hobject_t>, hobject_t>>(
-        std::make_tuple(objects, next.hobj));
-    });
+  auto gend = end.is_max() ? ghobject_t::get_max() : ghobject_t{end, 0, shard};
+  auto [gobjects, next] = co_await interruptor::make_interruptible(
+    store->list_objects(coll, gstart, gend, limit));
+
+  std::vector<hobject_t> objects;
+  boost::copy(
+    gobjects |
+    boost::adaptors::filtered([](const ghobject_t& o) {
+      if (o.is_pgmeta()) {
+	return false;
+      } else if (o.hobj.is_temp()) {
+	return false;
+      } else if (o.is_internal_pg_local()) {
+	return false;
+      } else {
+	return o.is_no_gen();
+      }
+    }) |
+    boost::adaptors::transformed([](const ghobject_t& o) {
+      return o.hobj;
+    }),
+    std::back_inserter(objects));
+  co_return std::make_tuple(objects, next.hobj);
 }
 
 PGBackend::setxattr_ierrorator::future<> PGBackend::setxattr(