osd/PGLog: skip ERROR entires in _merge_object_divergent_entries

During consistency check, do not take the version of ERROR entries
as the valid prior version of the following non-error entry.

Fixes: http://tracker.ceph.com/issues/20843
Signed-off-by: Jeegn Chen <jeegnchen@gmail.com>

diff --git a/src/osd/PGLog.h b/src/osd/PGLog.h
index 19405de25be67aac67edb054f9a7c6d9ecf94f2e..2cbe9cbc0e83d5bdd4996b3c8b2b7d8ed8ffb89f 100644 (file)
--- a/src/osd/PGLog.h
+++ b/src/osd/PGLog.h
@@ -840,23 +840,38 @@ protected:
     // strip out and ignore ERROR entries
     mempool::osd_pglog::list<pg_log_entry_t> entries;
     eversion_t last;
+    bool seen_non_error = false;
     for (list<pg_log_entry_t>::const_iterator i = orig_entries.begin();
         i != orig_entries.end();
         ++i) {
       // all entries are on hoid
       assert(i->soid == hoid);
-      if (i != orig_entries.begin() && i->prior_version != eversion_t()) {
+      // did not see error entries before this entry and this entry is not error
+      // then this entry is the first non error entry
+      bool first_non_error = ! seen_non_error && ! i->is_error();
+      if (! i->is_error() ) {
+        // see a non error entry now
+        seen_non_error = true;
+      }
+      
+      // No need to check the first entry since it prior_version is unavailable
+      // in the list
+      // No need to check if the prior_version is the minimal version
+      // No need to check the first non-error entry since the leading error
+      // entries are not its prior version
+      if (i != orig_entries.begin() && i->prior_version != eversion_t() &&
+          ! first_non_error) {
        // in increasing order of version
        assert(i->version > last);
        // prior_version correct (unless it is an ERROR entry)
        assert(i->prior_version == last || i->is_error());
       }
-      last = i->version;
       if (i->is_error()) {
        ldpp_dout(dpp, 20) << __func__ << ": ignoring " << *i << dendl;
       } else {
        ldpp_dout(dpp, 20) << __func__ << ": keeping " << *i << dendl;
        entries.push_back(*i);
+       last = i->version;
       }
     }
     if (entries.empty()) {

diff --git a/src/test/osd/TestPGLog.cc b/src/test/osd/TestPGLog.cc
index 70a71328a848d77d1300795ed4f2f42cb13766e9..bf29f87c4e55fda6d174835da9f36b3ade15f24e 100644 (file)
--- a/src/test/osd/TestPGLog.cc
+++ b/src/test/osd/TestPGLog.cc
@@ -93,6 +93,16 @@ struct PGLogTestBase {
     e.reqid = reqid;
     return e;
   }
+  static pg_log_entry_t mk_ple_err(
+    const hobject_t &hoid, eversion_t v, osd_reqid_t reqid) {
+    pg_log_entry_t e;
+    e.op = pg_log_entry_t::ERROR;
+    e.soid = hoid;
+    e.version = v;
+    e.prior_version = eversion_t(0, 0);
+    e.reqid = reqid;
+    return e;
+  }
   static pg_log_entry_t mk_ple_mod(
     const hobject_t &hoid, eversion_t v, eversion_t pv) {
     return mk_ple_mod(hoid, v, pv, osd_reqid_t());
@@ -109,6 +119,10 @@ struct PGLogTestBase {
     const hobject_t &hoid, eversion_t v, eversion_t pv) {
     return mk_ple_dt_rb(hoid, v, pv, osd_reqid_t());
   }
+  static pg_log_entry_t mk_ple_err(
+    const hobject_t &hoid, eversion_t v) {
+    return mk_ple_err(hoid, v, osd_reqid_t());
+  }
 }; // PGLogTestBase
 
 
@@ -2909,6 +2923,62 @@ TEST_F(PGLogTrimTest, TestGetRequest) {
   EXPECT_FALSE(result);
 }
 
+TEST_F(PGLogTest, _merge_object_divergent_entries) {
+  {
+    // Test for issue 20843
+    clear();
+    hobject_t hoid(object_t(/*name*/"notify.7"),
+                   /*key*/string(""),
+                   /*snap*/7,
+                   /*hash*/77,
+                   /*pool*/5,
+                   /*nspace*/string(""));
+    mempool::osd_pglog::list<pg_log_entry_t> orig_entries;
+    orig_entries.push_back(mk_ple_mod(hoid, eversion_t(8336, 957), eversion_t(8336, 952)));
+    orig_entries.push_back(mk_ple_err(hoid, eversion_t(8336, 958)));
+    orig_entries.push_back(mk_ple_err(hoid, eversion_t(8336, 959)));
+    orig_entries.push_back(mk_ple_mod(hoid, eversion_t(8336, 960), eversion_t(8336, 957)));
+    log.add(mk_ple_mod(hoid, eversion_t(8973, 1075), eversion_t(8971, 1070)));
+    missing.add(hoid,
+                /*need*/eversion_t(8971, 1070),
+                /*have*/eversion_t(8336, 952),
+                false);
+    pg_info_t oinfo;
+    LogHandler rollbacker;
+    _merge_object_divergent_entries(log, hoid,
+                                    orig_entries, oinfo,
+                                    log.get_can_rollback_to(),
+                                    missing, &rollbacker,
+                                    this);
+    // No core dump
+  }
+  {
+    // skip leading error entries
+    clear();
+    hobject_t hoid(object_t(/*name*/"notify.7"),
+                   /*key*/string(""),
+                   /*snap*/7,
+                   /*hash*/77,
+                   /*pool*/5,
+                   /*nspace*/string(""));
+    mempool::osd_pglog::list<pg_log_entry_t> orig_entries;
+    orig_entries.push_back(mk_ple_err(hoid, eversion_t(8336, 956)));
+    orig_entries.push_back(mk_ple_mod(hoid, eversion_t(8336, 957), eversion_t(8336, 952)));
+    log.add(mk_ple_mod(hoid, eversion_t(8973, 1075), eversion_t(8971, 1070)));
+    missing.add(hoid,
+                /*need*/eversion_t(8971, 1070),
+                /*have*/eversion_t(8336, 952),
+                false);
+    pg_info_t oinfo;
+    LogHandler rollbacker;
+    _merge_object_divergent_entries(log, hoid,
+                                    orig_entries, oinfo,
+                                    log.get_can_rollback_to(),
+                                    missing, &rollbacker,
+                                    this);
+    // No core dump
+  }
+}
 
 // Local Variables:
 // compile-command: "cd ../.. ; make unittest_pglog ; ./unittest_pglog --log-to-stderr=true  --debug-osd=20 # --gtest_filter=*.* "