Add superblock to KeyValueStore

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/os/KeyValueStore.cc b/src/os/KeyValueStore.cc
index d15a5c78118bac9a6653ef04c72177febca7201f..4f20ef6d778c5bccccddb01bb21638931953924d 100644 (file)
--- a/src/os/KeyValueStore.cc
+++ b/src/os/KeyValueStore.cc
@@ -70,6 +70,24 @@ const string KeyValueStore::OBJECT_OMAP_HEADER_KEY = "__OBJOMAP_HEADER__KEY_";
 const string KeyValueStore::COLLECTION = "__COLLECTION__";
 const string KeyValueStore::COLLECTION_ATTR = "__COLL_ATTR__";
 
+
+//Initial features in new superblock.
+static CompatSet get_kv_initial_compat_set() {
+  CompatSet::FeatureSet ceph_osd_feature_compat;
+  CompatSet::FeatureSet ceph_osd_feature_ro_compat;
+  CompatSet::FeatureSet ceph_osd_feature_incompat;
+  return CompatSet(ceph_osd_feature_compat, ceph_osd_feature_ro_compat,
+                  ceph_osd_feature_incompat);
+}
+
+//Features are added here that this KeyValueStore supports.
+static CompatSet get_kv_supported_compat_set() {
+  CompatSet compat =  get_kv_initial_compat_set();
+  //Any features here can be set in code, but not in initial superblock
+  return compat;
+}
+
+
 // ============== StripObjectMap Implementation =================
 
 int StripObjectMap::save_strip_header(StripObjectHeaderRef strip_header,
@@ -520,6 +538,8 @@ KeyValueStore::KeyValueStore(const std::string &base,
 
   g_ceph_context->get_perfcounters_collection()->add(perf_logger);
   g_ceph_context->_conf->add_observer(this);
+
+  superblock.compat_features = get_kv_initial_compat_set();
 }
 
 KeyValueStore::~KeyValueStore()
@@ -614,12 +634,20 @@ int KeyValueStore::mkfs()
     goto close_fsid_fd;
   }
 
+  // superblock
+  superblock.backend = g_conf->keyvaluestore_backend;
+  ret = write_superblock();
+  if (ret < 0) {
+    derr << "KeyValueStore::mkfs write_superblock() failed: "
+        << cpp_strerror(ret) << dendl;
+    goto close_fsid_fd;
+  }
+
   {
-    ret = KeyValueDB::test_init(g_conf->keyvaluestore_backend,
-                                              current_fn.c_str());
+    ret = KeyValueDB::test_init(superblock.backend, current_fn.c_str());
     if(ret < 0)
     {
-      derr << __func__ " failed to create backend type "
+      derr << __func__  << " failed to create backend type "
           << g_conf->keyvaluestore_backend << "." << dendl;
       ret = -1;
       goto close_fsid_fd;
@@ -693,6 +721,36 @@ bool KeyValueStore::test_mount_in_use()
   return inuse;
 }
 
+int KeyValueStore::write_superblock()
+{
+  bufferlist bl;
+  ::encode(superblock, bl);
+  return safe_write_file(basedir.c_str(), "superblock",
+      bl.c_str(), bl.length());
+}
+
+int KeyValueStore::read_superblock()
+{
+  bufferptr bp(PATH_MAX);
+  int ret = safe_read_file(basedir.c_str(), "superblock",
+      bp.c_str(), bp.length());
+  if (ret < 0) {
+    if (ret == -ENOENT) {
+      // If the file doesn't exist write initial CompatSet
+      return write_superblock();
+    }
+    return ret;
+  }
+
+  bufferlist bl;
+  bl.push_back(bp);
+  bufferlist::iterator i = bl.begin();
+  ::decode(superblock, i);
+  return 0;
+}
+
+
+
 int KeyValueStore::update_version_stamp()
 {
   return write_version_stamp();
@@ -731,6 +789,7 @@ int KeyValueStore::mount()
 {
   int ret;
   char buf[PATH_MAX];
+  CompatSet supported_compat_set = get_kv_supported_compat_set();
 
   dout(5) << "basedir " << basedir << dendl;
 
@@ -787,6 +846,20 @@ int KeyValueStore::mount()
       goto close_fsid_fd;
     }
   }
+  
+  ret = read_superblock();
+  if (ret < 0) {
+    ret = -EINVAL;
+    goto close_fsid_fd;
+  }
+
+  // Check if this KeyValueStore supports all the necessary features to mount
+  if (supported_compat_set.compare(superblock.compat_features) == -1) {
+    derr << "KeyValueStore::mount : Incompatible features set "
+          << superblock.compat_features << dendl;
+    ret = -EINVAL;
+    goto close_fsid_fd;
+  }
 
   current_fd = ::open(current_fn.c_str(), O_RDONLY);
   if (current_fd < 0) {
@@ -801,12 +874,12 @@ int KeyValueStore::mount()
   {
 
     KeyValueDB *store = KeyValueDB::create(g_ceph_context,
-                                          g_conf->keyvaluestore_backend,
+                                          superblock.backend,
                                           current_fn.c_str());
     if(!store)
     {
       derr << "KeyValueStore::mount backend type "
-          << g_conf->keyvaluestore_backend << " error" << dendl;
+          << superblock.backend << " error" << dendl;
       ret = -1;
       goto close_fsid_fd;
 
@@ -816,7 +889,7 @@ int KeyValueStore::mount()
     stringstream err;
     if (store->open(err)) {
       derr << "KeyValueStore::mount Error initializing keyvaluestore backend "
-           << g_conf->keyvaluestore_backend << ": " << err.str() << dendl;
+           << superblock.backend << ": " << err.str() << dendl;
       ret = -1;
       delete store;
       goto close_current_fd;
@@ -3026,3 +3099,44 @@ void KeyValueStore::handle_conf_change(const struct md_config_t *conf,
 void KeyValueStore::dump_transactions(list<ObjectStore::Transaction*>& ls, uint64_t seq, OpSequencer *osr)
 {
 }
+
+
+// -- KVSuperblock --
+
+void KVSuperblock::encode(bufferlist &bl) const
+{
+  ENCODE_START(1, 1, bl);
+  compat_features.encode(bl);
+  ::encode(backend, bl);
+  ENCODE_FINISH(bl);
+}
+
+void KVSuperblock::decode(bufferlist::iterator &bl)
+{
+  DECODE_START(1, bl);
+  compat_features.decode(bl);
+  ::decode(backend, bl);
+  DECODE_FINISH(bl);
+}
+
+void KVSuperblock::dump(Formatter *f) const
+{
+  f->open_object_section("compat");
+  compat_features.dump(f);
+  f->dump_string("backend", backend);
+  f->close_section();
+}
+
+void KVSuperblock::generate_test_instances(list<KVSuperblock*>& o)
+{
+  KVSuperblock z;
+  o.push_back(new KVSuperblock(z));
+  CompatSet::FeatureSet feature_compat;
+  CompatSet::FeatureSet feature_ro_compat;
+  CompatSet::FeatureSet feature_incompat;
+  z.compat_features = CompatSet(feature_compat, feature_ro_compat,
+                                feature_incompat);
+  o.push_back(new KVSuperblock(z));
+  z.backend = "rocksdb";
+  o.push_back(new KVSuperblock(z));
+}

diff --git a/src/os/KeyValueStore.h b/src/os/KeyValueStore.h
index 7bb15ca675738029d6143a940457efa2d0d25589..2d8ed0f1619cb41559b453db6b0413734c300154 100644 (file)
--- a/src/os/KeyValueStore.h
+++ b/src/os/KeyValueStore.h
@@ -137,6 +137,28 @@ class StripObjectMap: public GenericObjectMap {
 };
 
 
+class KVSuperblock {
+public:
+  CompatSet compat_features;
+  string backend;
+
+  KVSuperblock() { }
+
+  void encode(bufferlist &bl) const;
+  void decode(bufferlist::iterator &bl);
+  void dump(Formatter *f) const;
+  static void generate_test_instances(list<KVSuperblock*>& o);
+};
+WRITE_CLASS_ENCODER(KVSuperblock)
+
+
+inline ostream& operator<<(ostream& out, const KVSuperblock& sb)
+{
+  return out << "sb(" << sb.compat_features << "): "
+             << sb.backend;
+}
+
+
 class KeyValueStore : public ObjectStore,
                       public md_config_obs_t {
  public:
@@ -659,6 +681,25 @@ class KeyValueStore : public ObjectStore,
   static const string COLLECTION;
   static const string COLLECTION_ATTR;
   static const uint32_t COLLECTION_VERSION = 1;
+
+  KVSuperblock superblock;
+  /**
+   * write_superblock()
+   *
+   * Write superblock to persisent storage
+   *
+   * return value: 0 on success, otherwise negative errno
+   */
+  int write_superblock();
+
+  /**
+   * read_superblock()
+   *
+   * Fill in KeyValueStore::superblock by reading persistent storage
+   *
+   * return value: 0 on success, otherwise negative errno
+   */
+  int read_superblock();
 };
 
 WRITE_CLASS_ENCODER(StripObjectMap::StripObjectHeader)