ReplicatedPG: replace clean_up_local with a debug check

Stray objects should have been cleaned up in the merge_log
transactions.  Only on the primary have those operations
necessarily been flushed at activate().

Fixes: 5084
Signed-off-by: Samuel Just <sam.just@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>

5 files changed, 24 insertions, 13 deletions

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 94e737387c4..1d1377c72d5 100644
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -461,6 +461,7 @@ OPTION(osd_debug_drop_pg_create_duration, OPT_INT, 1)
 OPTION(osd_debug_drop_op_probability, OPT_DOUBLE, 0)   // probability of stalling/dropping a client op
 OPTION(osd_debug_op_order, OPT_BOOL, false)
 OPTION(osd_debug_verify_snaps_on_info, OPT_BOOL, false)
+OPTION(osd_debug_verify_stray_on_activate, OPT_BOOL, false)
 OPTION(osd_debug_skip_full_check_in_backfill_reservation, OPT_BOOL, false)
 OPTION(osd_op_history_size, OPT_U32, 20)    // Max number of completed ops to track
 OPTION(osd_op_history_duration, OPT_U32, 600) // Oldest completed op to track
diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 46c6f395997..0663b86665d 100644
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -1132,8 +1132,9 @@ void PG::activate(ObjectStore::Transaction& t,
   dirty_info = true;
   dirty_big_info = true; // maybe
 
-  // clean up stray objects
-  clean_up_local(t); 
+  // verify that there are no stray objects
+  if (is_primary())
+    check_local();
 
   // find out when we commit
   tfin.push_back(new C_PG_ActivateCommitted(this, query_epoch));
diff --git a/src/osd/PG.h b/src/osd/PG.h
index 2239e6b1275..e1f64dbc9fb 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -613,7 +613,7 @@ public:
     return pg_log.get_missing().num_missing() - missing_loc.size();
   }
 
-  virtual void clean_up_local(ObjectStore::Transaction& t) = 0;
+  virtual void check_local() = 0;
 
   virtual int start_recovery_ops(int max, RecoveryCtx *prctx) = 0;
 
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 918498edc55..14708e38cd9 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -7502,16 +7502,19 @@ void ReplicatedPG::scan_range(hobject_t begin, int min, int max, BackfillInterva
 }
 
 
-/** clean_up_local
- * remove any objects that we're storing but shouldn't.
- * as determined by log.
+/** check_local
+ * 
+ * verifies that stray objects have been deleted
  */
-void ReplicatedPG::clean_up_local(ObjectStore::Transaction& t)
+void ReplicatedPG::check_local()
 {
-  dout(10) << "clean_up_local" << dendl;
+  dout(10) << __func__ << dendl;
 
   assert(info.last_update >= pg_log.get_tail());  // otherwise we need some help!
 
+  if (!g_conf->osd_debug_verify_stray_on_activate)
+    return;
+
   // just scan the log.
   set<hobject_t> did;
   for (list<pg_log_entry_t>::const_reverse_iterator p = pg_log.get_log().log.rbegin();
@@ -7522,11 +7525,17 @@ void ReplicatedPG::clean_up_local(ObjectStore::Transaction& t)
     did.insert(p->soid);
 
     if (p->is_delete()) {
-      dout(10) << " deleting " << p->soid
-	       << " when " << p->version << dendl;
-      remove_snap_mapped_object(t, p->soid);
+      dout(10) << " checking " << p->soid
+	       << " at " << p->version << dendl;
+      struct stat st;
+      int r = osd->store->stat(coll, p->soid, &st);
+      if (r != -ENOENT) {
+	dout(10) << "Object " << p->soid << " exists, but should have been "
+		 << "deleted" << dendl;
+	assert(0 == "erroneously present object");
+      }
     } else {
-      // keep old(+missing) objects, just for kicks.
+      // ignore old(+missing) objects
     }
   }
 }
diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 4cedf913217..8f0cf6ec31c 100644
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -750,7 +750,7 @@ protected:
   int prepare_transaction(OpContext *ctx);
   
   // pg on-disk content
-  void clean_up_local(ObjectStore::Transaction& t);
+  void check_local();
 
   void _clear_recovery_state();