Merge pull request #4712 from dachary/wip-11663-verify-erasure-code-profile

normalize and verify the erasure code profile

Reviewed-by: Andreas Peters <andreas.joachim.peters@cern.ch>

src/erasure-code/ErasureCode.h

@@ -33,9 +33,19 @@ namespace ceph {
     static const unsigned SIMD_ALIGN;
 
     vector<int> chunk_mapping;
+    ErasureCodeProfile _profile;
 
     virtual ~ErasureCode() {}
 
+    virtual int init(ErasureCodeProfile &profile, ostream *ss) {
+      _profile = profile;
+      return 0;
+    }
+
+    virtual const ErasureCodeProfile &get_profile() const {
+      return _profile;
+    }
+
     virtual unsigned int get_coding_chunk_count() const {
       return get_chunk_count() - get_data_chunk_count();
     }

src/erasure-code/ErasureCodeInterface.h

@@ -153,6 +153,20 @@ using namespace std;
 namespace ceph {
 
   typedef map<std::string,std::string> ErasureCodeProfile;
+
+  inline ostream& operator<<(ostream& out, const ErasureCodeProfile& profile) {
+    out << "{";
+    for (ErasureCodeProfile::const_iterator it = profile.begin();
+	 it != profile.end();
+	 ++it) {
+      if (it != profile.begin()) out << ",";
+      out << it->first << "=" << it->second;
+    }
+    out << "}";
+    return out;
+  }
+
+
   class ErasureCodeInterface {
   public:
     virtual ~ErasureCodeInterface() {}
@@ -173,6 +187,14 @@ namespace ceph {
      */
     virtual int init(ErasureCodeProfile &profile, ostream *ss) = 0;
 
+    /**
+     * Return the profile that was used to initialize the instance
+     * with the **init** method.
+     *
+     * @return the profile in use by the instance
+     */
+    virtual const ErasureCodeProfile &get_profile() const = 0;
+
     /**
      * Create a new ruleset in **crush** under the name **name**,
      * unless it already exists.

src/erasure-code/ErasureCodePlugin.cc

@@ -102,7 +102,15 @@ int ErasureCodePluginRegistry::factory(const std::string &plugin_name,
     }
   }
 
-  return plugin->factory(profile, erasure_code, ss);
+  int r = plugin->factory(profile, erasure_code, ss);
+  if (r)
+    return r;
+  if (profile != (*erasure_code)->get_profile()) {
+    *ss << __func__ << " profile " << profile << " != get_profile() "
+	<< (*erasure_code)->get_profile() << std::endl;
+    return -EINVAL;
+  }
+  return 0;
 }
 
 static const char *an_older_version() {

src/erasure-code/isa/ErasureCodeIsa.cc

@@ -80,6 +80,7 @@ ErasureCodeIsa::init(ErasureCodeProfile &profile, ostream *ss)
   if (err)
     return err;
   prepare();
+  ErasureCode::init(profile, ss);
   return err;
 }
 

src/erasure-code/jerasure/ErasureCodeJerasure.cc

@@ -67,6 +67,7 @@ int ErasureCodeJerasure::init(ErasureCodeProfile& profile, ostream *ss)
   if (err)
     return err;
   prepare();
+  ErasureCode::init(profile, ss);
   return err;
 }
 

src/erasure-code/lrc/ErasureCodeLrc.cc

@@ -534,6 +534,7 @@ int ErasureCodeLrc::init(ErasureCodeProfile &profile,
     profile.erase("mapping");
     profile.erase("layers");
   }
+  ErasureCode::init(profile, ss);
   return 0;
 }
 

src/erasure-code/shec/ErasureCodeShec.cc

@@ -68,6 +68,7 @@ int ErasureCodeShec::init(ErasureCodeProfile &profile,
   if (err)
     return err;
   prepare();
+  ErasureCode::init(profile, ss);
   return err;
 }
 

src/mon/OSDMonitor.cc

@@ -3940,6 +3940,17 @@ int OSDMonitor::crush_rename_bucket(const string& srcname,
   return 0;
 }
 
+int OSDMonitor::normalize_profile(ErasureCodeProfile &profile, ostream *ss)
+{
+  ErasureCodeInterfaceRef erasure_code;
+  ErasureCodePluginRegistry &instance = ErasureCodePluginRegistry::instance();
+  ErasureCodeProfile::const_iterator plugin = profile.find("plugin");
+  int err = instance.factory(plugin->second, profile, &erasure_code, ss);
+  if (err)
+    return err;
+  return erasure_code->init(profile, ss);
+}
+
 int OSDMonitor::crush_ruleset_create_erasure(const string &name,
 					     const string &profile,
 					     int *ruleset,
@@ -5452,22 +5463,6 @@ bool OSDMonitor::prepare_command_impl(MMonCommand *m,
     }
     string plugin = profile_map["plugin"];
 
-    if (osdmap.has_erasure_code_profile(name)) {
-      if (osdmap.get_erasure_code_profile(name) == profile_map) {
-	err = 0;
-	goto reply;
-      }
-      if (!force) {
-	err = -EPERM;
-	ss << "will not override erasure code profile " << name
-	   << " because the existing profile "
-	   << osdmap.get_erasure_code_profile(name)
-	   << " is different from the proposed profile "
-	   << profile_map;
-	goto reply;
-      }
-    }
-
     if (pending_inc.has_erasure_code_profile(name)) {
       dout(20) << "erasure code profile " << name << " try again" << dendl;
       goto wait;
@@ -5484,7 +5479,34 @@ bool OSDMonitor::prepare_command_impl(MMonCommand *m,
 	  goto reply;
 	}
       }
-      dout(20) << "erasure code profile " << name << " set" << dendl;
+      err = normalize_profile(profile_map, &ss);
+      if (err)
+	goto reply;
+
+      if (osdmap.has_erasure_code_profile(name)) {
+	ErasureCodeProfile existing_profile_map =
+	  osdmap.get_erasure_code_profile(name);
+	err = normalize_profile(existing_profile_map, &ss);
+	if (err)
+	  goto reply;
+
+	if (existing_profile_map == profile_map) {
+	  err = 0;
+	  goto reply;
+	}
+	if (!force) {
+	  err = -EPERM;
+	  ss << "will not override erasure code profile " << name
+	     << " because the existing profile "
+	     << existing_profile_map
+	     << " is different from the proposed profile "
+	     << profile_map;
+	  goto reply;
+	}
+      }
+
+      dout(20) << "erasure code profile set " << name << "="
+	       << profile_map << dendl;
       pending_inc.set_erasure_code_profile(name, profile_map);
     }
 
@@ -5518,7 +5540,11 @@ bool OSDMonitor::prepare_command_impl(MMonCommand *m,
 						      &ss);
 	if (err)
 	  goto reply;
-	dout(20) << "erasure code profile " << profile << " set" << dendl;
+	err = normalize_profile(profile_map, &ss);
+	if (err)
+	  goto reply;
+	dout(20) << "erasure code profile set " << profile << "="
+		 << profile_map << dendl;
 	pending_inc.set_erasure_code_profile(profile, profile_map);
 	goto wait;
       }

src/mon/OSDMonitor.h

@@ -277,6 +277,7 @@ private:
   int crush_rename_bucket(const string& srcname,
 			  const string& dstname,
 			  ostream *ss);
+  int normalize_profile(ErasureCodeProfile &profile, ostream *ss);
   int crush_ruleset_create_erasure(const string &name,
 				   const string &profile,
 				   int *ruleset,

src/test/erasure-code/ErasureCodeExample.h

@@ -39,10 +39,6 @@ class ErasureCodeExample : public ErasureCode {
 public:
   virtual ~ErasureCodeExample() {}
 
-  virtual int init(ErasureCodeProfile &profile, ostream *ss) {
-    return 0;
-  }
-
   virtual int create_ruleset(const string &name,
 			     CrushWrapper &crush,
 			     ostream *ss) const {

src/test/erasure-code/ErasureCodePluginExample.cc

@@ -28,6 +28,7 @@ public:
 		      ostream *ss)
   {
     *erasure_code = ErasureCodeInterfaceRef(new ErasureCodeExample());
+    (*erasure_code)->init(profile, ss);
     return 0;
   }
 };

src/test/erasure-code/Makefile.am

@@ -35,7 +35,9 @@ ceph_erasure_code_LDADD += -ldl
 endif
 bin_DEBUGPROGRAMS += ceph_erasure_code
 
-libec_example_la_SOURCES = test/erasure-code/ErasureCodePluginExample.cc
+libec_example_la_SOURCES = \
+	erasure-code/ErasureCode.cc \
+	test/erasure-code/ErasureCodePluginExample.cc
 test/erasure-code/ErasureCodePluginExample.cc: ./ceph_ver.h
 libec_example_la_CFLAGS = ${AM_CFLAGS}
 libec_example_la_CXXFLAGS= ${AM_CXXFLAGS}

src/test/mon/osd-crush.sh

@@ -113,7 +113,7 @@ function TEST_crush_rule_create_erasure() {
     ! ./ceph osd erasure-code-profile ls | grep default || return 1
     ./ceph osd crush rule create-erasure $ruleset || return 1
     CEPH_ARGS='' ./ceph --admin-daemon $dir/ceph-mon.a.asok log flush || return 1
-    grep 'profile default set' $dir/mon.a.log || return 1
+    grep 'profile set default' $dir/mon.a.log || return 1
     ./ceph osd erasure-code-profile ls | grep default || return 1
     ./ceph osd crush rule rm $ruleset || return 1
     ! ./ceph osd crush rule ls | grep $ruleset || return 1