osd: Add digest of omap for deep-scrub

Add ScrubMap encode/decode v4 message with omap digest
Compute digest of header and key/value.  Use bufferlist
to reflect structure and compute as we go, clearing
bufferlist to reduce memory usage.

Signed-off-by: David Zafman <david.zafman@inktank.com>
Reviewed-by: Samuel Just <sam.just@inktank.com>

src/osd/PG.cc

@@ -3162,8 +3162,8 @@ void PG::_scan_list(ScrubMap &map, vector<hobject_t> &ls, bool deep)
 
       // calculate the CRC32 on deep scrubs
       if (deep) {
-        bufferhash h;
-        bufferlist bl;
+        bufferhash h, oh;
+        bufferlist bl, hdrbl;
         int r;
         __u64 pos = 0;
         while ( (r = osd->store->read(coll, poid, pos,
@@ -3174,6 +3174,33 @@ void PG::_scan_list(ScrubMap &map, vector<hobject_t> &ls, bool deep)
         }
         o.digest = h.digest();
         o.digest_present = true;
+
+        bl.clear();
+        r = osd->store->omap_get_header(coll, poid, &hdrbl);
+        if (r == 0) {
+          dout(25) << "CRC header " << string(hdrbl.c_str(), hdrbl.length())
+             << dendl;
+          ::encode(hdrbl, bl);
+          oh << bl;
+          bl.clear();
+        }
+
+        ObjectMap::ObjectMapIterator iter = osd->store->get_omap_iterator(
+          coll, poid);
+        assert(iter);
+        for (iter->seek_to_first(); iter->valid() ; iter->next()) {
+          dout(25) << "CRC key " << iter->key() << " value "
+            << string(iter->value().c_str(), iter->value().length()) << dendl;
+
+          ::encode(iter->key(), bl);
+          ::encode(iter->value(), bl);
+          oh << bl;
+          bl.clear();
+        }
+
+        //Store final calculated CRC32 of omap header & key/values
+        o.omap_digest = oh.digest();
+        o.omap_digest_present = true;
       }
 
       if (poid.snap != CEPH_SNAPDIR && poid.snap != CEPH_NOSNAP) {
@@ -4076,6 +4103,16 @@ bool PG::_compare_scrub_objects(ScrubMap::object &auth,
                   << " != known digest " << auth.digest;
     }
   }
+  if (auth.omap_digest_present && candidate.omap_digest_present) {
+    if (auth.omap_digest != candidate.omap_digest) {
+      if (!ok)
+        errorstream << ", ";
+      ok = false;
+
+      errorstream << "omap_digest " << candidate.omap_digest
+                  << " != known omap_digest " << auth.omap_digest;
+    }
+  }
   for (map<string,bufferptr>::const_iterator i = auth.attrs.begin();
        i != auth.attrs.end();
        i++) {

src/osd/osd_types.cc

@@ -2756,7 +2756,7 @@ void ScrubMap::generate_test_instances(list<ScrubMap*>& o)
 
 void ScrubMap::object::encode(bufferlist& bl) const
 {
-  ENCODE_START(4, 2, bl);
+  ENCODE_START(5, 2, bl);
   ::encode(size, bl);
   ::encode(negative, bl);
   ::encode(attrs, bl);
@@ -2764,12 +2764,14 @@ void ScrubMap::object::encode(bufferlist& bl) const
   ::encode(digest_present, bl);
   ::encode(nlinks, bl);
   ::encode(snapcolls, bl);
+  ::encode(omap_digest, bl);
+  ::encode(omap_digest_present, bl);
   ENCODE_FINISH(bl);
 }
 
 void ScrubMap::object::decode(bufferlist::iterator& bl)
 {
-  DECODE_START_LEGACY_COMPAT_LEN(4, 2, 2, bl);
+  DECODE_START_LEGACY_COMPAT_LEN(5, 2, 2, bl);
   ::decode(size, bl);
   ::decode(negative, bl);
   ::decode(attrs, bl);
@@ -2777,10 +2779,6 @@ void ScrubMap::object::decode(bufferlist::iterator& bl)
     ::decode(digest, bl);
     ::decode(digest_present, bl);
   }
-  else {
-    digest = 0;
-    digest_present = false;
-  }
   if (struct_v >= 4) {
     ::decode(nlinks, bl);
     ::decode(snapcolls, bl);
@@ -2789,6 +2787,10 @@ void ScrubMap::object::decode(bufferlist::iterator& bl)
      * return nlink >= 1 */
     nlinks = 0;
   }
+  if (struct_v >= 5) {
+    ::decode(omap_digest, bl);
+    ::decode(omap_digest_present, bl);
+  }
   DECODE_FINISH(bl);
 }
 

src/osd/osd_types.h

@@ -1831,10 +1831,12 @@ struct ScrubMap {
     bool digest_present;
     uint32_t nlinks;
     set<snapid_t> snapcolls;
+    __u32 omap_digest;
+    bool omap_digest_present;
 
     object() :
       size(0), negative(false), digest(0), digest_present(false),
-      nlinks(0) {}
+      nlinks(0), omap_digest(0), omap_digest_present(false) {}
 
     void encode(bufferlist& bl) const;
     void decode(bufferlist::iterator& bl);