Merge pull request ceph#27200 from neha-ojha/wip-21174-2

osd/PGLog: preserve original_crt to check rollbackability

Reviewed-by: xie xingguo <[email protected]>

src/osd/PGLog.cc

@@ -269,13 +269,19 @@ void PGLog::proc_replica_log(
     limit :
     first_non_divergent->version;
 
+  // We need to preserve the original crt before it gets updated in rewind_from_head().
+  // Later, in merge_object_divergent_entries(), we use it to check whether we can rollback
+  // a divergent entry or not.
+  eversion_t original_crt = log.get_can_rollback_to();
+  dout(20) << __func__ << " original_crt = " << original_crt << dendl;
   IndexedLog folog(olog);
   auto divergent = folog.rewind_from_head(lu);
   _merge_divergent_entries(
     folog,
     divergent,
     oinfo,
     olog.get_can_rollback_to(),
+    original_crt,
     omissing,
     0,
     this);
@@ -318,7 +324,11 @@ void PGLog::rewind_divergent_log(eversion_t newhead,
   dout(10) << "rewind_divergent_log truncate divergent future " <<
     newhead << dendl;
 
-
+  // We need to preserve the original crt before it gets updated in rewind_from_head().
+  // Later, in merge_object_divergent_entries(), we use it to check whether we can rollback
+  // a divergent entry or not.
+  eversion_t original_crt = log.get_can_rollback_to();
+  dout(20) << __func__ << " original_crt = " << original_crt << dendl;
   if (info.last_complete > newhead)
     info.last_complete = newhead;
 
@@ -336,6 +346,7 @@ void PGLog::rewind_divergent_log(eversion_t newhead,
     divergent,
     info,
     log.get_can_rollback_to(),
+    original_crt,
     missing,
     rollbacker,
     this);
@@ -433,6 +444,11 @@ void PGLog::merge_log(pg_info_t &oinfo, pg_log_t &olog, pg_shard_t fromosd,
 	     << lower_bound << dendl;
     mark_dirty_from(lower_bound);
 
+    // We need to preserve the original crt before it gets updated in rewind_from_head().
+    // Later, in merge_object_divergent_entries(), we use it to check whether we can rollback
+    // a divergent entry or not.
+    eversion_t original_crt = log.get_can_rollback_to();
+    dout(20) << __func__ << " original_crt = " << original_crt << dendl;
     auto divergent = log.rewind_from_head(lower_bound);
     // move aside divergent items
     for (auto &&oe: divergent) {
@@ -457,6 +473,7 @@ void PGLog::merge_log(pg_info_t &oinfo, pg_log_t &olog, pg_shard_t fromosd,
       divergent,
       info,
       log.get_can_rollback_to(),
+      original_crt,
       missing,
       rollbacker,
       this);

src/osd/PGLog.h

@@ -867,6 +867,7 @@ struct PGLog : DoutPrefixProvider {
     const mempool::osd_pglog::list<pg_log_entry_t> &orig_entries, ///< [in] entries for hoid to merge
     const pg_info_t &info,              ///< [in] info for merging entries
     eversion_t olog_can_rollback_to,     ///< [in] rollback boundary
+    eversion_t original_can_rollback_to,     ///< [in] original rollback boundary
     missing_type &missing,               ///< [in,out] missing to adjust, use
     LogEntryHandler *rollbacker,         ///< [in] optional rollbacker object
     const DoutPrefixProvider *dpp        ///< [in] logging provider
@@ -930,6 +931,8 @@ struct PGLog : DoutPrefixProvider {
     const bool object_not_in_store =
       !missing.is_missing(hoid) &&
       entries.rbegin()->is_delete();
+    ldpp_dout(dpp, 10) << __func__ << ": hoid " << " object_not_in_store: "
+                       << object_not_in_store << dendl;
     ldpp_dout(dpp, 10) << __func__ << ": hoid " << hoid
 		       << " prior_version: " << prior_version
 		       << " first_divergent_update: " << first_divergent_update
@@ -1028,11 +1031,17 @@ struct PGLog : DoutPrefixProvider {
     ldpp_dout(dpp, 10) << __func__ << ": hoid " << hoid
                        << " olog_can_rollback_to: "
                        << olog_can_rollback_to << dendl;
+    ldpp_dout(dpp, 10) << __func__ << ": hoid " << hoid
+                       << " original_crt: "
+                       << original_can_rollback_to << dendl;
     /// Distinguish between 4) and 5)
     for (list<pg_log_entry_t>::const_reverse_iterator i = entries.rbegin();
 	 i != entries.rend();
 	 ++i) {
-      if (!i->can_rollback() || i->version <= olog_can_rollback_to) {
+      /// Use original_can_rollback_to instead of olog_can_rollback_to to check
+      //  if we can rollback or not. This is to ensure that we don't try to rollback
+      //  to an object that has been deleted and doesn't exist.
+      if (!i->can_rollback() || i->version <= original_can_rollback_to) {
 	ldpp_dout(dpp, 10) << __func__ << ": hoid " << hoid << " cannot rollback "
 			   << *i << dendl;
 	can_rollback = false;
@@ -1045,7 +1054,7 @@ struct PGLog : DoutPrefixProvider {
       for (list<pg_log_entry_t>::const_reverse_iterator i = entries.rbegin();
 	   i != entries.rend();
 	   ++i) {
-	ceph_assert(i->can_rollback() && i->version > olog_can_rollback_to);
+	ceph_assert(i->can_rollback() && i->version > original_can_rollback_to);
 	ldpp_dout(dpp, 10) << __func__ << ": hoid " << hoid
 			   << " rolling back " << *i << dendl;
 	if (rollbacker)
@@ -1082,6 +1091,7 @@ struct PGLog : DoutPrefixProvider {
     mempool::osd_pglog::list<pg_log_entry_t> &entries,       ///< [in] entries to merge
     const pg_info_t &oinfo,              ///< [in] info for merging entries
     eversion_t olog_can_rollback_to,     ///< [in] rollback boundary
+    eversion_t original_can_rollback_to, ///< [in] original rollback boundary
     missing_type &omissing,              ///< [in,out] missing to adjust, use
     LogEntryHandler *rollbacker,         ///< [in] optional rollbacker object
     const DoutPrefixProvider *dpp        ///< [in] logging provider
@@ -1097,6 +1107,7 @@ struct PGLog : DoutPrefixProvider {
 	i->second,
 	oinfo,
 	olog_can_rollback_to,
+        original_can_rollback_to,
 	omissing,
 	rollbacker,
 	dpp);
@@ -1120,6 +1131,7 @@ struct PGLog : DoutPrefixProvider {
       entries,
       info,
       log.get_can_rollback_to(),
+      log.get_can_rollback_to(),
       missing,
       rollbacker,
       this);

src/test/osd/TestPGLog.cc

@@ -2939,6 +2939,7 @@ TEST_F(PGLogTest, _merge_object_divergent_entries) {
     _merge_object_divergent_entries(log, hoid,
                                     orig_entries, oinfo,
                                     log.get_can_rollback_to(),
+                                    log.get_can_rollback_to(),
                                     missing, &rollbacker,
                                     this);
     // No core dump
@@ -2965,6 +2966,7 @@ TEST_F(PGLogTest, _merge_object_divergent_entries) {
     _merge_object_divergent_entries(log, hoid,
                                     orig_entries, oinfo,
                                     log.get_can_rollback_to(),
+                                    log.get_can_rollback_to(),
                                     missing, &rollbacker,
                                     this);
     // No core dump