Merge branch 'wip-3651'

8 files changed, 222 insertions, 9 deletions

diff --git a/src/common/ceph_argparse.cc b/src/common/ceph_argparse.cc
index 646931acdc9..f26fe93b656 100644
--- a/src/common/ceph_argparse.cc
+++ b/src/common/ceph_argparse.cc
@@ -45,6 +45,17 @@
 #undef generic_dout
 #undef dendl
 
+void string_to_vec(std::vector<std::string>& args, std::string argstr)
+{
+  istringstream iss(argstr);
+  while(iss) {
+    string sub;
+    iss >> sub;
+    if (sub == "") break;
+    args.push_back(sub);
+  }
+}
+
 void env_to_vec(std::vector<const char*>& args, const char *name)
 {
   if (!name)
diff --git a/src/common/ceph_argparse.h b/src/common/ceph_argparse.h
index fd9439c5627..3ef0251abeb 100644
--- a/src/common/ceph_argparse.h
+++ b/src/common/ceph_argparse.h
@@ -43,6 +43,7 @@ public:
 };
 
 /////////////////////// Functions ///////////////////////
+extern void string_to_vec(std::vector<std::string>& args, std::string argstr);
 extern void env_to_vec(std::vector<const char*>& args, const char *name=NULL);
 extern void argv_to_vec(int argc, const char **argv,
                  std::vector<const char*>& args);
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 4549eb76671..34ec3678f08 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -17,6 +17,7 @@
 #include <errno.h>
 #include <sys/stat.h>
 #include <signal.h>
+#include <ctype.h>
 #include <boost/scoped_ptr.hpp>
 
 #if defined(DARWIN) || defined(__FreeBSD__)
@@ -869,6 +870,22 @@ public:
   }
 };
 
+class TestOpsSocketHook : public AdminSocketHook {
+  OSDService *service;
+  ObjectStore *store;
+public:
+  TestOpsSocketHook(OSDService *s, ObjectStore *st) : service(s), store(st) {}
+  bool call(std::string command, std::string args, bufferlist& out) {
+    stringstream ss;
+    test_ops(service, store, command, args, ss);
+    out.append(ss);
+    return true;
+  }
+  void test_ops(OSDService *service, ObjectStore *store, std::string command,
+     std::string args, ostream &ss);
+
+};
+
 int OSD::init()
 {
   Mutex::Locker lock(osd_lock);
@@ -966,10 +983,24 @@ int OSD::init()
   AdminSocket *admin_socket = cct->get_admin_socket();
   r = admin_socket->register_command("dump_ops_in_flight", admin_ops_hook,
                                          "show the ops currently in flight");
+  assert(r == 0);
   historic_ops_hook = new HistoricOpsSocketHook(this);
   r = admin_socket->register_command("dump_historic_ops", historic_ops_hook,
                                          "show slowest recent ops");
   assert(r == 0);
+  test_ops_hook = new TestOpsSocketHook(&(this->service), this->store);
+  r = admin_socket->register_command("setomapval", test_ops_hook,
+                              "setomap <pool-id> <obj-name> <key> <val>");
+  assert(r == 0);
+  r = admin_socket->register_command("rmomapkey", test_ops_hook,
+                               "rmomapkey <pool-id> <obj-name> <key>");
+  assert(r == 0);
+  r = admin_socket->register_command("setomapheader", test_ops_hook,
+                               "setomapheader <pool-id> <obj-name> <header>");
+  assert(r == 0);
+  r = admin_socket->register_command("getomap", test_ops_hook,
+                               "getomap <pool-id> <obj-name>");
+  assert(r == 0);
 
   service.init();
   service.publish_map(osdmap);
@@ -1120,10 +1151,17 @@ int OSD::shutdown()
   dout(10) << "no ops" << dendl;
 
   cct->get_admin_socket()->unregister_command("dump_ops_in_flight");
+  cct->get_admin_socket()->unregister_command("dump_historic_ops");
   delete admin_ops_hook;
   delete historic_ops_hook;
   admin_ops_hook = NULL;
   historic_ops_hook = NULL;
+  cct->get_admin_socket()->unregister_command("setomapval");
+  cct->get_admin_socket()->unregister_command("rmomapkey");
+  cct->get_admin_socket()->unregister_command("setomapheader");
+  cct->get_admin_socket()->unregister_command("getomap");
+  delete test_ops_hook;
+  test_ops_hook = NULL;
 
   recovery_tp.stop();
   dout(10) << "recovery tp stopped" << dendl;
@@ -2291,6 +2329,120 @@ void OSD::dump_ops_in_flight(ostream& ss)
   op_tracker.dump_ops_in_flight(ss);
 }
 
+// Usage:
+//   setomapval <pool-id> <obj-name> <key> <val>
+//   rmomapkey <pool-id> <obj-name> <key>
+//   setomapheader <pool-id> <obj-name> <header>
+void TestOpsSocketHook::test_ops(OSDService *service, ObjectStore *store,
+     std::string command, std::string args, ostream &ss)
+{
+  //Test support
+  //Support changing the omap on a single osd by using the Admin Socket to
+  //directly request the osd make a change.
+  if (command == "setomapval" || command == "rmomapkey" ||
+        command == "setomapheader" || command == "getomap") {
+    std::vector<std::string> argv;
+    pg_t rawpg, pgid;
+    int64_t pool;
+    OSDMapRef curmap = service->get_osdmap();
+    int r;
+
+    argv.push_back(command);
+    string_to_vec(argv, args);
+    int argc = argv.size();
+
+    if (argc < 3) {
+      ss << "Illegal request";
+      return;
+    }
+ 
+    pool = curmap->const_lookup_pg_pool_name(argv[1].c_str());
+    //If we can't find it my name then maybe id specified
+    if (pool < 0 && isdigit(argv[1].c_str()[0]))
+      pool = atoll(argv[1].c_str());
+    r = -1;
+    if (pool >= 0)
+        r = curmap->object_locator_to_pg(object_t(argv[2]),
+          object_locator_t(pool), rawpg);
+    if (r < 0) {
+        ss << "Invalid pool " << argv[1];
+        return;
+    }
+    pgid = curmap->raw_pg_to_pg(rawpg);
+
+    hobject_t obj(object_t(argv[2]), string(""), CEPH_NOSNAP, rawpg.ps(), pool);
+    ObjectStore::Transaction t;
+
+    if (command == "setomapval") {
+      if (argc != 5) {
+        ss << "usage: setomapval <pool> <obj-name> <key> <val>";
+        return;
+      }
+      map<string, bufferlist> newattrs;
+      bufferlist val;
+      string key(argv[3]);
+ 
+      val.append(argv[4]);
+      newattrs[key] = val;
+      t.omap_setkeys(coll_t(pgid), obj, newattrs);
+      r = store->apply_transaction(t);
+      if (r < 0)
+        ss << "error=" << r;
+      else
+        ss << "ok";
+    } else if (command == "rmomapkey") {
+      if (argc != 4) {
+        ss << "usage: rmomapkey <pool> <obj-name> <key>";
+        return;
+      }
+      set<string> keys;
+
+      keys.insert(string(argv[3]));
+      t.omap_rmkeys(coll_t(pgid), obj, keys);
+      r = store->apply_transaction(t);
+      if (r < 0)
+        ss << "error=" << r;
+      else
+        ss << "ok";
+    } else if (command == "setomapheader") {
+      if (argc != 4) {
+        ss << "usage: setomapheader <pool> <obj-name> <header>";
+        return;
+      }
+      bufferlist newheader;
+
+      newheader.append(argv[3]);
+      t.omap_setheader(coll_t(pgid), obj, newheader);
+      r = store->apply_transaction(t);
+      if (r < 0)
+        ss << "error=" << r;
+      else
+        ss << "ok";
+    } else if (command == "getomap") {
+      if (argc != 3) {
+        ss << "usage: getomap <pool> <obj-name>";
+        return;
+      }
+      //Debug: Output entire omap
+      bufferlist hdrbl;
+      map<string, bufferlist> keyvals;
+      r = store->omap_get(coll_t(pgid), obj, &hdrbl, &keyvals);
+      if (r >= 0) {
+          ss << "header=" << string(hdrbl.c_str(), hdrbl.length());
+          for (map<string, bufferlist>::iterator it = keyvals.begin();
+              it != keyvals.end(); it++)
+            ss << " key=" << (*it).first << " val="
+               << string((*it).second.c_str(), (*it).second.length());
+      } else {
+          ss << "error=" << r;
+      }
+    }
+    return;
+  }
+  ss << "Internal error - command=" << command;
+  return;
+}
+
 // =========================================
 void OSD::RemoveWQ::_process(boost::tuple<coll_t, SequencerRef, DeletingStateRef> *item)
 {
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 83308486f0f..be4ed11791e 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -133,6 +133,7 @@ class AuthAuthorizeHandlerRegistry;
 
 class OpsFlightSocketHook;
 class HistoricOpsSocketHook;
+class TestOpsSocketHook;
 struct C_CompleteSplits;
 
 extern const coll_t meta_coll;
@@ -617,11 +618,14 @@ private:
   void dump_historic_ops(ostream& ss) {
     return op_tracker.dump_historic_ops(ss);
   }
+  void test_ops(std::string command, std::string args, ostream& ss);
   friend class OpsFlightSocketHook;
   friend class HistoricOpsSocketHook;
+  friend class TestOpsSocketHook;
   friend class C_CompleteSplits;
   OpsFlightSocketHook *admin_ops_hook;
   HistoricOpsSocketHook *historic_ops_hook;
+  TestOpsSocketHook *test_ops_hook;
 
   // -- op queue --
 
diff --git a/src/osd/OSDMap.h b/src/osd/OSDMap.h
index 4d4bbd0b031..5105fc7ab0e 100644
--- a/src/osd/OSDMap.h
+++ b/src/osd/OSDMap.h
@@ -450,6 +450,10 @@ public:
     return -ENOENT;
   }
 
+  int64_t const_lookup_pg_pool_name(const char *name) const {
+    return const_cast<OSDMap *>(this)->lookup_pg_pool_name(name);
+  }
+
   int64_t get_pool_max() const {
     return pool_max;
   }
diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 32d5830cd21..2899738c346 100644
--- a/src/osd/PG.cc
+++ b/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++) {
diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index b98613ccd43..008462af426 100644
--- a/src/osd/osd_types.cc
+++ b/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);
 }
 
diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index 95859404c19..5fc9e83912f 100644
--- a/src/osd/osd_types.h
+++ b/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);