Merge pull request ceph#58173 from ronen-fr/wip-rf-targets-j13

osd/scrub: no shared scrub-job ownership between PGs and the scrub queue

Reviewed-by: Samuel Just <[email protected]>

Author: ronen-fr

qa/standalone/scrub/osd-scrub-test.sh

@@ -540,8 +540,8 @@ function TEST_dump_scrub_schedule() {
             --osd_scrub_interval_randomize_ratio=0 \
             --osd_scrub_backoff_ratio=0.0 \
             --osd_op_queue=wpq \
-            --osd_stats_update_period_not_scrubbing=3 \
-            --osd_stats_update_period_scrubbing=2 \
+            --osd_stats_update_period_not_scrubbing=1 \
+            --osd_stats_update_period_scrubbing=1 \
             --osd_scrub_sleep=0.2"
 
     for osd in $(seq 0 $(expr $OSDS - 1))
@@ -562,7 +562,8 @@ function TEST_dump_scrub_schedule() {
     rm -f $TESTDATA
 
     local pgid="${poolid}.0"
-    local now_is=`date -I"ns"`
+    #local now_is=`date -I"ns"` # note: uses a comma for the ns part
+    local now_is=`date +'%Y-%m-%dT%H:%M:%S.%N%:z'`
 
     # before the scrubbing starts
 
@@ -604,8 +605,8 @@ function TEST_dump_scrub_schedule() {
     #         scheduled for the future' value
     #
 
-    ceph tell osd.* config set osd_scrub_chunk_max "3" || return 1
-    ceph tell osd.* config set osd_scrub_sleep "1.0" || return 1
+    ceph tell osd.* config set osd_shallow_scrub_chunk_max "3" || return 1
+    ceph tell osd.* config set osd_scrub_sleep "2.0" || return 1
     ceph osd set noscrub || return 1
     sleep 2
     saved_last_stamp=${sched_data['query_last_stamp']}

qa/standalone/scrub/scrub-helpers.sh

@@ -18,7 +18,7 @@ function extract_published_sch() {
   local -n dict=$4 # a ref to the in/out dictionary
   local current_time=$2
   local extra_time=$3
-  local extr_dbg=1 # note: 3 and above leave some temp files around
+  local extr_dbg=2 # note: 3 and above leave some temp files around
 
   #turn off '-x' (but remember previous state)
   local saved_echo_flag=${-//[^x]/}
@@ -51,18 +51,26 @@ function extract_published_sch() {
   (( extr_dbg >= 2 )) && echo "query output:"
   (( extr_dbg >= 2 )) && ceph pg $1 query -f json-pretty | awk -e '/scrubber/,/agent_state/ {print;}'
 
+  # note: the query output for the schedule containas two dates: the first is the not-before, and
+  # the second is the original target time (which is before or the same as the not-before)
+  # the current line format looks like this:
+  # "schedule": "scrub scheduled @ 2024-06-26T16:09:56.666 (2024-06-24T16:09:56.338)"
   from_qry=`ceph pg $1 query -f json-pretty | jq -r --arg extra_dt "$extra_time" --arg current_dt "$current_time"  --arg spt "'" '
     . |
         (.q_stat_part=((.scrubber.schedule// "-") | if test(".*@.*") then (split(" @ ")|first) else . end)) |
         (.q_when_part=((.scrubber.schedule// "0") | if test(".*@.*") then (split(" @ ")|last) else "0" end)) |
-	(.q_when_is_future=(.q_when_part > $current_dt)) |
+        (.q_target=((.scrubber.schedule// "0") | if test(".*@.*") then (split(" @ ")|last|split(" (")|last|split(")")|first) else "0" end)) |
+        (.q_not_before=((.scrubber.schedule// "0") | if test(".*@.*") then (split(" @ ")|last|split(" (")|first) else "0" end)) |
+	(.q_when_is_future=(.q_target > $current_dt)) |
 	(.q_vs_date=(.q_when_part > $extra_dt)) |	
       {
         query_epoch: .epoch,
         query_seq: .info.stats.reported_seq,
         query_active: (.scrubber | if has("active") then .active else "bug" end),
         query_schedule: .q_stat_part,
-        query_schedule_at: .q_when_part,
+        #query_schedule_at: .q_when_part,
+        query_schedule_at: .q_not_before,
+        query_target_at: .q_target,
         query_last_duration: .info.stats.last_scrub_duration,
         query_last_stamp: .info.history.last_scrub_stamp,
         query_last_scrub: (.info.history.last_scrub| sub($spt;"x") ),

src/osd/PG.cc

@@ -1325,6 +1325,7 @@ unsigned int PG::scrub_requeue_priority(Scrub::scrub_prio_t with_priority, unsig
 
 
 Scrub::schedule_result_t PG::start_scrubbing(
+    std::unique_ptr<Scrub::ScrubJob> candidate,
     Scrub::OSDRestrictions osd_restrictions)
 {
   dout(10) << fmt::format(
@@ -1348,10 +1349,9 @@ Scrub::schedule_result_t PG::start_scrubbing(
        get_pgbackend()->auto_repair_supported());
 
   return m_scrubber->start_scrub_session(
-      osd_restrictions, pg_cond, m_planned_scrub);
+      std::move(candidate), osd_restrictions, pg_cond, m_planned_scrub);
 }
 
-
 double PG::next_deepscrub_interval() const
 {
   double deep_scrub_interval =
@@ -1360,14 +1360,22 @@ double PG::next_deepscrub_interval() const
     deep_scrub_interval = cct->_conf->osd_deep_scrub_interval;
   return info.history.last_deep_scrub_stamp + deep_scrub_interval;
 }
-void PG::on_scrub_schedule_input_change()
+
+void PG::on_scrub_schedule_input_change(Scrub::delay_ready_t delay_ready)
 {
-  if (is_active() && is_primary()) {
-    dout(20) << __func__ << ": active/primary" << dendl;
+  if (is_active() && is_primary() && !is_scrub_queued_or_active()) {
+    dout(10) << fmt::format(
+		    "{}: active/primary. delay_ready={:c}", __func__,
+		    (delay_ready == Scrub::delay_ready_t::delay_ready) ? 't'
+								       : 'f')
+	     << dendl;
     ceph_assert(m_scrubber);
-    m_scrubber->update_scrub_job(m_planned_scrub);
+    m_scrubber->update_scrub_job(delay_ready);
   } else {
-    dout(20) << __func__ << ": inactive or non-primary" << dendl;
+    dout(10) << fmt::format(
+		    "{}: inactive, non-primary - or already scrubbing",
+		    __func__)
+	     << dendl;
   }
 }
 
@@ -2260,7 +2268,7 @@ void PG::handle_activate_map(PeeringCtx &rctx, epoch_t range_starts_at)
   // on_scrub_schedule_input_change() as pool.info contains scrub scheduling
   // parameters.
   if (pool.info.last_change >= range_starts_at) {
-    on_scrub_schedule_input_change();
+    on_scrub_schedule_input_change(Scrub::delay_ready_t::delay_ready);
   }
 }
 

src/osd/PG.h

@@ -269,7 +269,7 @@ class PG : public DoutPrefixProvider,
 	set_last_scrub_stamp(t, history, stats);
 	return true;
       });
-    on_scrub_schedule_input_change();
+    on_scrub_schedule_input_change(Scrub::delay_ready_t::delay_ready);
   }
 
   static void set_last_deep_scrub_stamp(
@@ -285,7 +285,7 @@ class PG : public DoutPrefixProvider,
 	set_last_scrub_stamp(t, history, stats);
 	return true;
       });
-    on_scrub_schedule_input_change();
+    on_scrub_schedule_input_change(Scrub::delay_ready_t::delay_ready);
   }
 
   static void add_objects_scrubbed_count(
@@ -531,7 +531,7 @@ class PG : public DoutPrefixProvider,
    * - pg stat scrub timestamps
    * - etc
    */
-  void on_scrub_schedule_input_change();
+  void on_scrub_schedule_input_change(Scrub::delay_ready_t delay_ready);
 
   void scrub_requested(scrub_level_t scrub_level, scrub_type_t scrub_type) override;
 
@@ -701,7 +701,8 @@ class PG : public DoutPrefixProvider,
   bool get_must_scrub() const;
 
   Scrub::schedule_result_t start_scrubbing(
-      Scrub::OSDRestrictions osd_restrictions);
+    std::unique_ptr<Scrub::ScrubJob> candidate,
+    Scrub::OSDRestrictions osd_restrictions);
 
   unsigned int scrub_requeue_priority(
       Scrub::scrub_prio_t with_priority,

src/osd/osd_types_fmt.h

@@ -9,6 +9,7 @@
 #include "osd/osd_types.h"
 #include <fmt/chrono.h>
 #include <fmt/ranges.h>
+#include <fmt/std.h>
 #if FMT_VERSION >= 90000
 #include <fmt/ostream.h>
 #endif

src/osd/scrubber/osd_scrub.cc

@@ -49,7 +49,12 @@ OsdScrub::~OsdScrub()
 
 std::ostream& OsdScrub::gen_prefix(std::ostream& out, std::string_view fn) const
 {
-  return out << m_log_prefix << fn << ": ";
+  if (fn.starts_with("operator")) {
+    // it's a lambda, and __func__ is not available
+    return out << m_log_prefix;
+  } else {
+    return out << m_log_prefix << fn << ": ";
+  }
 }
 
 void OsdScrub::dump_scrubs(ceph::Formatter* f) const
@@ -82,15 +87,16 @@ bool OsdScrub::scrub_random_backoff() const
   return false;
 }
 
+void OsdScrub::debug_log_all_jobs() const
+{
+  m_queue.for_each_job([this](const Scrub::ScrubJob& sj) {
+    dout(20) << fmt::format("\tscrub-queue jobs: {}", sj) << dendl;
+  }, 20);
+}
+
 
 void OsdScrub::initiate_scrub(bool is_recovery_active)
 {
-  const utime_t scrub_time = ceph_clock_now();
-  dout(10) << fmt::format(
-		  "time now:{:s}, recovery is active?:{}", scrub_time,
-		  is_recovery_active)
-	   << dendl;
-
   if (auto blocked_pgs = get_blocked_pgs_count(); blocked_pgs > 0) {
     // some PGs managed by this OSD were blocked by a locked object during
     // scrub. This means we might not have the resources needed to scrub now.
@@ -101,53 +107,44 @@ void OsdScrub::initiate_scrub(bool is_recovery_active)
 	<< dendl;
   }
 
+  const utime_t scrub_time = ceph_clock_now();
+
   // check the OSD-wide environment conditions (scrub resources, time, etc.).
   // These may restrict the type of scrubs we are allowed to start, or just
   // prevent us from starting any non-operator-initiated scrub at all.
-  auto env_restrictions =
+  const auto env_restrictions =
       restrictions_on_scrubbing(is_recovery_active, scrub_time);
 
+  dout(10) << fmt::format("scrub scheduling (@tick) starts. "
+                          "time now:{:s}, recovery is active?:{} restrictions:{}",
+                          scrub_time, is_recovery_active, env_restrictions)
+	   << dendl;
+
   if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>() &&
       !env_restrictions.high_priority_only) {
-    dout(20) << "scrub scheduling (@tick) starts" << dendl;
-    auto all_jobs = m_queue.list_registered_jobs();
-    for (const auto& sj : all_jobs) {
-      dout(20) << fmt::format("\tscrub-queue jobs: {}", *sj) << dendl;
-    }
+    debug_log_all_jobs();
   }
 
-  // at this phase of the refactoring: minimal changes to the
-  // queue interface used here: we ask for a list of
-  // eligible targets (based on the known restrictions).
-  // We try all elements of this list until a (possibly temporary) success.
-  auto candidates = m_queue.ready_to_scrub(env_restrictions, scrub_time);
-  if (candidates.empty()) {
+  auto candidate = m_queue.pop_ready_pg(env_restrictions, scrub_time);
+  if (!candidate) {
     dout(20) << "no PGs are ready for scrubbing" << dendl;
     return;
   }
 
-  for (const auto& candidate : candidates) {
-    dout(20) << fmt::format("initiating scrub on pg[{}]", candidate) << dendl;
-
-    // we have a candidate to scrub. But we may fail when trying to initiate that
-    // scrub. For some failures - we can continue with the next candidate. For
-    // others - we should stop trying to scrub at this tick.
-    auto res = initiate_a_scrub(candidate, env_restrictions);
+  auto candidate_pg = candidate->pgid;
+  auto res = initiate_a_scrub(std::move(candidate), env_restrictions);
 
-    if (res == schedule_result_t::target_specific_failure) {
-      // continue with the next job.
-      // \todo: consider separate handling of "no such PG", as - later on -
-      // we should be removing both related targets.
-      continue;
-    } else if (res == schedule_result_t::osd_wide_failure) {
-      // no point in trying the other candidates at this time
+  switch (res) {
+    case schedule_result_t::target_specific_failure:
+    case schedule_result_t::osd_wide_failure:
+      // No scrub this tick.
+      // someone else will requeue the target, if needed.
       break;
-    } else {
-      // the happy path. We are done
-      dout(20) << fmt::format("scrub initiated for pg[{}]", candidate.pgid)
-               << dendl;
+
+    case schedule_result_t::scrub_initiated:
+      dout(20) << fmt::format("scrub initiated for pg[{}]", candidate_pg)
+	       << dendl;
       break;
-    }
   }
 }
 
@@ -198,45 +195,52 @@ Scrub::OSDRestrictions OsdScrub::restrictions_on_scrubbing(
 
 
 Scrub::schedule_result_t OsdScrub::initiate_a_scrub(
-    spg_t pgid,
+    std::unique_ptr<Scrub::ScrubJob> candidate,
     Scrub::OSDRestrictions restrictions)
 {
-  dout(20) << fmt::format("trying pg[{}]", pgid) << dendl;
+  dout(20) << fmt::format("trying pg[{}]", candidate->pgid) << dendl;
 
   // we have a candidate to scrub. We need some PG information to
   // know if scrubbing is allowed
 
-  auto locked_pg = m_osd_svc.get_locked_pg(pgid);
+  auto locked_pg = m_osd_svc.get_locked_pg(candidate->pgid);
   if (!locked_pg) {
-    // the PG was dequeued in the short timespan between creating the
-    // candidates list (ready_to_scrub()) and here
-    dout(5) << fmt::format("pg[{}] not found", pgid) << dendl;
+    // the PG was dequeued in the short timespan between querying the
+    // scrub queue - and now.
+    dout(5) << fmt::format("pg[{}] not found", candidate->pgid) << dendl;
     return Scrub::schedule_result_t::target_specific_failure;
   }
 
-  // later on, here is where the scrub target would be dequeued
-  return locked_pg->pg()->start_scrubbing(restrictions);
+  // note: the 'candidate', which in this step is a copy of the scrub job,
+  // was already dequeued. The "original" scrub job cannot be accessed from
+  // here directly. Thus - we leave it to start_scrubbing() (via a call
+  // to PgScrubber::start_scrub_session() to mark it as dequeued.
+  return locked_pg->pg()->start_scrubbing(std::move(candidate), restrictions);
 }
 
+
 void OsdScrub::on_config_change()
 {
-  auto to_notify = m_queue.list_registered_jobs();
+  auto to_notify = m_queue.get_pgs([](const Scrub::ScrubJob& sj) -> bool {
+    ceph_assert(sj.registered);
+    return true;
+  });
 
   for (const auto& p : to_notify) {
-    dout(30) << fmt::format("rescheduling pg[{}] scrubs", *p) << dendl;
-    auto locked_pg = m_osd_svc.get_locked_pg(p->pgid);
+    dout(30) << fmt::format("rescheduling pg[{}] scrubs", p) << dendl;
+    auto locked_pg = m_osd_svc.get_locked_pg(p);
     if (!locked_pg)
       continue;
 
     dout(15) << fmt::format(
 		    "updating scrub schedule on {}",
 		    (locked_pg->pg())->get_pgid())
 	     << dendl;
-    locked_pg->pg()->on_scrub_schedule_input_change();
+    locked_pg->pg()->on_scrub_schedule_input_change(
+	Scrub::delay_ready_t::no_delay);
   }
 }
 
-
 // ////////////////////////////////////////////////////////////////////////// //
 // CPU load tracking and related
 
@@ -421,34 +425,15 @@ PerfCounters* OsdScrub::get_perf_counters(int pool_type, scrub_level_t level)
 // ////////////////////////////////////////////////////////////////////////// //
 // forwarders to the queue
 
-void OsdScrub::update_job(
-    Scrub::ScrubJobRef sjob,
-    const Scrub::sched_params_t& suggested,
-    bool reset_notbefore)
-{
-  m_queue.update_job(sjob, suggested, reset_notbefore);
-}
-
-void OsdScrub::delay_on_failure(
-      Scrub::ScrubJobRef sjob,
-      std::chrono::seconds delay,
-      Scrub::delay_cause_t delay_cause,
-      utime_t now_is)
-{
-  m_queue.delay_on_failure(sjob, delay, delay_cause, now_is);
-}
-
 
-void OsdScrub::register_with_osd(
-    Scrub::ScrubJobRef sjob,
-    const Scrub::sched_params_t& suggested)
+void OsdScrub::enqueue_target(const Scrub::ScrubJob& sjob)
 {
-  m_queue.register_with_osd(sjob, suggested);
+  m_queue.enqueue_target(sjob);
 }
 
-void OsdScrub::remove_from_osd_queue(Scrub::ScrubJobRef sjob)
+void OsdScrub::remove_from_osd_queue(spg_t pgid)
 {
-  m_queue.remove_from_osd_queue(sjob);
+  m_queue.remove_from_osd_queue(pgid);
 }
 
 std::unique_ptr<Scrub::LocalResourceWrapper> OsdScrub::inc_scrubs_local(